wasmtime_wasi_http/
default_send_request.rs

1use crate::{Error, RequestOptions};
2use bytes::Bytes;
3use core::future::poll_fn;
4use core::pin::{Pin, pin};
5use core::task::{Poll, ready};
6use http::uri::Scheme;
7use http::{Request, Response};
8use http_body::Body;
9use std::time::Duration;
10use tokio::io::{AsyncRead, AsyncWrite};
11use tokio::net::TcpStream;
12
13trait TokioStream: AsyncRead + AsyncWrite + Send + Sync + Unpin + 'static {
14    fn boxed(self) -> Box<dyn TokioStream>
15    where
16        Self: Sized,
17    {
18        Box::new(self)
19    }
20}
21impl<T> TokioStream for T where T: AsyncRead + AsyncWrite + Send + Sync + Unpin + 'static {}
22
23/// The default implementation of how an outgoing request is sent.
24///
25/// This implementation is used by the `wasi:http/handler` interface
26/// default implementation.
27///
28/// The returned [Future] can be used to communicate
29/// a request processing error, if any, to the constructor of the request.
30/// For example, if the request was constructed via `wasi:http/types.request#new`,
31/// a result resolved from it will be forwarded to the guest on the future handle returned.
32///
33/// This function performs no `Content-Length` validation.
34pub async fn default_send_request(
35    mut req: Request<impl Body<Data = Bytes, Error = Error> + Send + 'static>,
36    options: Option<RequestOptions>,
37) -> Result<
38    (
39        Response<impl Body<Data = Bytes, Error = Error>>,
40        impl Future<Output = Result<(), Error>> + Send,
41    ),
42    Error,
43> {
44    let uri = req.uri();
45    let authority = uri.authority().ok_or(Error::HttpRequestUriInvalid)?;
46    let use_tls = uri.scheme() == Some(&Scheme::HTTPS);
47    let authority = if authority.port().is_some() {
48        authority.to_string()
49    } else {
50        let port = if use_tls { 443 } else { 80 };
51        format!("{authority}:{port}")
52    };
53
54    let connect_timeout = options
55        .and_then(|r| r.connect_timeout)
56        .unwrap_or(Duration::from_secs(600));
57
58    let first_byte_timeout = options
59        .and_then(|r| r.first_byte_timeout)
60        .unwrap_or(Duration::from_secs(600));
61
62    let between_bytes_timeout = options
63        .and_then(|r| r.between_bytes_timeout)
64        .unwrap_or(Duration::from_secs(600));
65
66    // `connect_timeout` covers everything up to a usable connection: the TCP
67    // connect, the TLS handshake, and the HTTP handshake.
68    let connect = async {
69        let stream = TcpStream::connect(&authority)
70            .await
71            .map_err(Error::Connect)?;
72        let stream = if use_tls {
73            // derived from https://github.com/rustls/rustls/blob/main/examples/src/bin/simpleclient.rs
74            let root_cert_store = rustls::RootCertStore {
75                roots: webpki_roots::TLS_SERVER_ROOTS.into(),
76            };
77            let config = rustls::ClientConfig::builder()
78                .with_root_certificates(root_cert_store)
79                .with_no_client_auth();
80            let connector = tokio_rustls::TlsConnector::from(std::sync::Arc::new(config));
81            let domain = tls_server_name(&authority)?;
82            let stream = connector
83                .connect(domain, stream)
84                .await
85                .map_err(Error::Tls)?;
86            stream.boxed()
87        } else {
88            stream.boxed()
89        };
90        // TODO: we should plumb the builder through the http context, and use it here
91        hyper::client::conn::http1::Builder::new()
92            .handshake(crate::io::TokioIo::new(stream))
93            .await
94            .map_err(Error::from)
95    };
96    let (mut sender, conn) = tokio::time::timeout(connect_timeout, connect)
97        .await
98        .map_err(|_| Error::ConnectionTimeout)??;
99
100    // at this point, the request contains the scheme and the authority, but
101    // the http packet should only include those if addressing a proxy, so
102    // remove them here, since SendRequest::send_request does not do it for us
103    *req.uri_mut() = http::Uri::builder()
104        .path_and_query(
105            req.uri()
106                .path_and_query()
107                .map(|p| p.as_str())
108                .unwrap_or("/"),
109        )
110        .build()
111        .expect("comes from valid request");
112
113    let send = async move {
114        use core::task::Context;
115
116        /// Wrapper around [hyper::body::Incoming] used to
117        /// account for request option timeout configuration
118        struct IncomingResponseBody {
119            incoming: hyper::body::Incoming,
120            timeout: tokio::time::Interval,
121        }
122        impl http_body::Body for IncomingResponseBody {
123            type Data = <hyper::body::Incoming as http_body::Body>::Data;
124            type Error = Error;
125
126            fn poll_frame(
127                mut self: Pin<&mut Self>,
128                cx: &mut Context<'_>,
129            ) -> Poll<Option<Result<http_body::Frame<Self::Data>, Self::Error>>> {
130                match Pin::new(&mut self.as_mut().incoming).poll_frame(cx) {
131                    Poll::Ready(None) => Poll::Ready(None),
132                    Poll::Ready(Some(Err(err))) => {
133                        let err = if err.is_timeout() {
134                            Error::HttpResponseTimeout
135                        } else {
136                            Error::from(err)
137                        };
138                        Poll::Ready(Some(Err(err)))
139                    }
140                    Poll::Ready(Some(Ok(frame))) => {
141                        self.timeout.reset();
142                        Poll::Ready(Some(Ok(frame)))
143                    }
144                    Poll::Pending => {
145                        ready!(self.timeout.poll_tick(cx));
146                        Poll::Ready(Some(Err(Error::ConnectionReadTimeout)))
147                    }
148                }
149            }
150            fn is_end_stream(&self) -> bool {
151                self.incoming.is_end_stream()
152            }
153            fn size_hint(&self) -> http_body::SizeHint {
154                self.incoming.size_hint()
155            }
156        }
157
158        let res = tokio::time::timeout(first_byte_timeout, sender.send_request(req))
159            .await
160            .map_err(|_| Error::ConnectionReadTimeout)?
161            .map_err(Error::from)?;
162        // Tokio panics on an `interval` with a zero duration, so clamp it to a
163        // 1ns timeout which is sort of effectively zero anyway.
164        let between_bytes_timeout = if between_bytes_timeout == Duration::ZERO {
165            Duration::new(0, 1)
166        } else {
167            between_bytes_timeout
168        };
169        let mut timeout = tokio::time::interval(between_bytes_timeout);
170        timeout.reset();
171        Ok(res.map(|incoming| IncomingResponseBody { incoming, timeout }))
172    };
173    let mut send = pin!(send);
174    let mut conn = Some(conn);
175    // Wait for response while driving connection I/O
176    let res = poll_fn(|cx| match send.as_mut().poll(cx) {
177        Poll::Ready(Ok(res)) => Poll::Ready(Ok(res)),
178        Poll::Ready(Err(err)) => Poll::Ready(Err(err)),
179        Poll::Pending => {
180            // Response is not ready, poll `hyper` connection to drive I/O if it has not completed yet
181            let Some(fut) = conn.as_mut() else {
182                // `hyper` connection already completed
183                return Poll::Pending;
184            };
185            let res = ready!(Pin::new(fut).poll(cx));
186            // `hyper` connection completed, record that to prevent repeated poll
187            conn = None;
188            match res {
189                // `hyper` connection has successfully completed, optimistically poll for response
190                Ok(()) => send.as_mut().poll(cx),
191                // `hyper` connection has failed, return the error
192                Err(err) => Poll::Ready(Err(Error::from(err))),
193            }
194        }
195    })
196    .await?;
197    Ok((res, async move {
198        let Some(conn) = conn.take() else {
199            // `hyper` connection has already completed
200            return Ok(());
201        };
202        if let Err(err) = conn.await {
203            if err.is_timeout() {
204                return Err(Error::HttpResponseTimeout);
205            }
206            return Err(err.into());
207        }
208        Ok(())
209    }))
210}
211
212/// Resolve the rustls [`ServerName`] used for TLS certificate verification from
213/// an outbound request `authority`.
214///
215/// `authority` is in `host:port` form, where an IPv6 `host` is wrapped in
216/// brackets (for example `[::1]:443`). An IP literal is recognized by parsing
217/// the whole authority as a [`SocketAddr`]; this handles the bracketed IPv6
218/// form, which splitting on the first `:` would truncate. Anything else is
219/// treated as a host name, with the port stripped off before it is handed to
220/// rustls.
221///
222/// [`ServerName`]: rustls::pki_types::ServerName
223/// [`SocketAddr`]: std::net::SocketAddr
224pub(crate) fn tls_server_name(
225    authority: &str,
226) -> Result<rustls::pki_types::ServerName<'static>, rustls::pki_types::InvalidDnsNameError> {
227    use rustls::pki_types::ServerName;
228
229    if let Ok(addr) = authority.parse::<std::net::SocketAddr>() {
230        return Ok(ServerName::from(addr.ip()));
231    }
232    let host = match authority.split_once(':') {
233        Some((host, _port)) => host,
234        None => authority,
235    };
236    Ok(ServerName::try_from(host)?.to_owned())
237}
238
239#[cfg(test)]
240mod tls_server_name_tests {
241    use super::tls_server_name;
242    use rustls::pki_types::ServerName;
243
244    #[test]
245    fn resolves_server_name_from_authority() {
246        // Host names keep their host and drop the port.
247        assert_eq!(
248            tls_server_name("example.com:443").unwrap(),
249            ServerName::try_from("example.com").unwrap()
250        );
251        assert_eq!(
252            tls_server_name("example.com").unwrap(),
253            ServerName::try_from("example.com").unwrap()
254        );
255
256        // IP literals resolve to an `IpAddress` server name. The bracketed IPv6
257        // form must not be truncated at the first `:`.
258        assert_eq!(
259            tls_server_name("127.0.0.1:80").unwrap(),
260            ServerName::from(std::net::Ipv4Addr::LOCALHOST)
261        );
262        assert_eq!(
263            tls_server_name("[::1]:443").unwrap(),
264            ServerName::from(std::net::Ipv6Addr::LOCALHOST)
265        );
266        assert_eq!(
267            tls_server_name("[2001:db8::1]:8443").unwrap(),
268            ServerName::from("2001:db8::1".parse::<std::net::Ipv6Addr>().unwrap())
269        );
270    }
271}
wasmtime_wasi_http/default_send_request.rs

wasmtime_wasi_http/
default_send_request.rs
