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    let stream = match tokio::time::timeout(connect_timeout, TcpStream::connect(&authority)).await {
67        Ok(stream) => stream.map_err(Error::Connect)?,
68        Err(..) => return Err(Error::ConnectionTimeout),
69    };
70    let stream = if use_tls {
71        // derived from https://github.com/rustls/rustls/blob/main/examples/src/bin/simpleclient.rs
72        let root_cert_store = rustls::RootCertStore {
73            roots: webpki_roots::TLS_SERVER_ROOTS.into(),
74        };
75        let config = rustls::ClientConfig::builder()
76            .with_root_certificates(root_cert_store)
77            .with_no_client_auth();
78        let connector = tokio_rustls::TlsConnector::from(std::sync::Arc::new(config));
79        let domain = tls_server_name(&authority)?;
80        let stream = connector
81            .connect(domain, stream)
82            .await
83            .map_err(Error::Tls)?;
84        stream.boxed()
85    } else {
86        stream.boxed()
87    };
88    let (mut sender, conn) = tokio::time::timeout(
89        connect_timeout,
90        // TODO: we should plumb the builder through the http context, and use it here
91        hyper::client::conn::http1::Builder::new().handshake(crate::io::TokioIo::new(stream)),
92    )
93    .await
94    .map_err(|_| Error::ConnectionTimeout)??;
95
96    // at this point, the request contains the scheme and the authority, but
97    // the http packet should only include those if addressing a proxy, so
98    // remove them here, since SendRequest::send_request does not do it for us
99    *req.uri_mut() = http::Uri::builder()
100        .path_and_query(
101            req.uri()
102                .path_and_query()
103                .map(|p| p.as_str())
104                .unwrap_or("/"),
105        )
106        .build()
107        .expect("comes from valid request");
108
109    let send = async move {
110        use core::task::Context;
111
112        /// Wrapper around [hyper::body::Incoming] used to
113        /// account for request option timeout configuration
114        struct IncomingResponseBody {
115            incoming: hyper::body::Incoming,
116            timeout: tokio::time::Interval,
117        }
118        impl http_body::Body for IncomingResponseBody {
119            type Data = <hyper::body::Incoming as http_body::Body>::Data;
120            type Error = Error;
121
122            fn poll_frame(
123                mut self: Pin<&mut Self>,
124                cx: &mut Context<'_>,
125            ) -> Poll<Option<Result<http_body::Frame<Self::Data>, Self::Error>>> {
126                match Pin::new(&mut self.as_mut().incoming).poll_frame(cx) {
127                    Poll::Ready(None) => Poll::Ready(None),
128                    Poll::Ready(Some(Err(err))) => {
129                        let err = if err.is_timeout() {
130                            Error::HttpResponseTimeout
131                        } else {
132                            Error::from(err)
133                        };
134                        Poll::Ready(Some(Err(err)))
135                    }
136                    Poll::Ready(Some(Ok(frame))) => {
137                        self.timeout.reset();
138                        Poll::Ready(Some(Ok(frame)))
139                    }
140                    Poll::Pending => {
141                        ready!(self.timeout.poll_tick(cx));
142                        Poll::Ready(Some(Err(Error::ConnectionReadTimeout)))
143                    }
144                }
145            }
146            fn is_end_stream(&self) -> bool {
147                self.incoming.is_end_stream()
148            }
149            fn size_hint(&self) -> http_body::SizeHint {
150                self.incoming.size_hint()
151            }
152        }
153
154        let res = tokio::time::timeout(first_byte_timeout, sender.send_request(req))
155            .await
156            .map_err(|_| Error::ConnectionReadTimeout)?
157            .map_err(Error::from)?;
158        let mut timeout = tokio::time::interval(between_bytes_timeout);
159        timeout.reset();
160        Ok(res.map(|incoming| IncomingResponseBody { incoming, timeout }))
161    };
162    let mut send = pin!(send);
163    let mut conn = Some(conn);
164    // Wait for response while driving connection I/O
165    let res = poll_fn(|cx| match send.as_mut().poll(cx) {
166        Poll::Ready(Ok(res)) => Poll::Ready(Ok(res)),
167        Poll::Ready(Err(err)) => Poll::Ready(Err(err)),
168        Poll::Pending => {
169            // Response is not ready, poll `hyper` connection to drive I/O if it has not completed yet
170            let Some(fut) = conn.as_mut() else {
171                // `hyper` connection already completed
172                return Poll::Pending;
173            };
174            let res = ready!(Pin::new(fut).poll(cx));
175            // `hyper` connection completed, record that to prevent repeated poll
176            conn = None;
177            match res {
178                // `hyper` connection has successfully completed, optimistically poll for response
179                Ok(()) => send.as_mut().poll(cx),
180                // `hyper` connection has failed, return the error
181                Err(err) => Poll::Ready(Err(Error::from(err))),
182            }
183        }
184    })
185    .await?;
186    Ok((res, async move {
187        let Some(conn) = conn.take() else {
188            // `hyper` connection has already completed
189            return Ok(());
190        };
191        if let Err(err) = conn.await {
192            if err.is_timeout() {
193                return Err(Error::HttpResponseTimeout);
194            }
195            return Err(err.into());
196        }
197        Ok(())
198    }))
199}
200
201/// Resolve the rustls [`ServerName`] used for TLS certificate verification from
202/// an outbound request `authority`.
203///
204/// `authority` is in `host:port` form, where an IPv6 `host` is wrapped in
205/// brackets (for example `[::1]:443`). An IP literal is recognized by parsing
206/// the whole authority as a [`SocketAddr`]; this handles the bracketed IPv6
207/// form, which splitting on the first `:` would truncate. Anything else is
208/// treated as a host name, with the port stripped off before it is handed to
209/// rustls.
210///
211/// [`ServerName`]: rustls::pki_types::ServerName
212/// [`SocketAddr`]: std::net::SocketAddr
213pub(crate) fn tls_server_name(
214    authority: &str,
215) -> Result<rustls::pki_types::ServerName<'static>, rustls::pki_types::InvalidDnsNameError> {
216    use rustls::pki_types::ServerName;
217
218    if let Ok(addr) = authority.parse::<std::net::SocketAddr>() {
219        return Ok(ServerName::from(addr.ip()));
220    }
221    let host = match authority.split_once(':') {
222        Some((host, _port)) => host,
223        None => authority,
224    };
225    Ok(ServerName::try_from(host)?.to_owned())
226}
227
228#[cfg(test)]
229mod tls_server_name_tests {
230    use super::tls_server_name;
231    use rustls::pki_types::ServerName;
232
233    #[test]
234    fn resolves_server_name_from_authority() {
235        // Host names keep their host and drop the port.
236        assert_eq!(
237            tls_server_name("example.com:443").unwrap(),
238            ServerName::try_from("example.com").unwrap()
239        );
240        assert_eq!(
241            tls_server_name("example.com").unwrap(),
242            ServerName::try_from("example.com").unwrap()
243        );
244
245        // IP literals resolve to an `IpAddress` server name. The bracketed IPv6
246        // form must not be truncated at the first `:`.
247        assert_eq!(
248            tls_server_name("127.0.0.1:80").unwrap(),
249            ServerName::from(std::net::Ipv4Addr::LOCALHOST)
250        );
251        assert_eq!(
252            tls_server_name("[::1]:443").unwrap(),
253            ServerName::from(std::net::Ipv6Addr::LOCALHOST)
254        );
255        assert_eq!(
256            tls_server_name("[2001:db8::1]:8443").unwrap(),
257            ServerName::from("2001:db8::1".parse::<std::net::Ipv6Addr>().unwrap())
258        );
259    }
260}
wasmtime_wasi_http/default_send_request.rs

wasmtime_wasi_http/
default_send_request.rs
