[48.0.0] Backports for security fixes (#14175)

* Limit buffered writes in http/files

This commit adds limits to the amount of data buffered from a guest on
the host when guests write to WASIp3 streams for files and http bodies.
This ensures that the guest can't control how much is allocated on the
host, for example, but rather it's limited to a fixed amount.

Co-authored-by: Till Schneidereit <till@tillschneidereit.net>

* Update cap-std dependencies

* Fix MSRV

---------

Co-authored-by: Till Schneidereit <till@tillschneidereit.net>

Author: alexcrichton

Cargo.lock

@@ -367,8 +367,8 @@ wasip1 = { version = "1.0.0", default-features = false }
 # Note that `cap-fs-ext` should be avoided where possible to use
 # `cap-primitives` instead.
 target-lexicon = "0.13.5"
-cap-primitives = "4.0.2"
-cap-fs-ext-avoid-using-this = { version = "4.0.2", package = 'cap-fs-ext' }
+cap-primitives = "4.0.3"
+cap-fs-ext-avoid-using-this = { version = "4.0.3", package = 'cap-fs-ext' }
 rustix = "1.1.4"
 # wit-bindgen:
 wit-bindgen = { version = "0.60.0", default-features = false }

Cargo.toml

@@ -0,0 +1,85 @@
+use futures::join;
+use test_programs::p3::wasi::filesystem::types::{
+    Descriptor, DescriptorFlags, OpenFlags, PathFlags,
+};
+use test_programs::p3::{wasi, wit_stream};
+use wit_bindgen::StreamResult;
+
+struct Component;
+
+test_programs::p3::export!(Component);
+
+impl test_programs::p3::exports::wasi::cli::run::Guest for Component {
+    async fn run() -> Result<(), ()> {
+        let preopens = wasi::filesystem::preopens::get_directories();
+        let (dir, _) = &preopens[0];
+        test_chunked_write(dir, "chunked_write.txt").await;
+        Ok(())
+    }
+}
+
+fn bytes(offset: &mut usize, len: usize) -> Vec<u8> {
+    let mut buf = Vec::with_capacity(len);
+    for i in 0..len {
+        buf.push(((*offset + i) % 251) as u8);
+    }
+    *offset += len;
+    buf
+}
+
+async fn test_chunked_write(dir: &Descriptor, filename: &str) {
+    let mut len = 16;
+    let mut pos = 0;
+
+    let file = dir
+        .open_at(
+            PathFlags::empty(),
+            filename.to_string(),
+            OpenFlags::CREATE,
+            DescriptorFlags::READ | DescriptorFlags::WRITE,
+        )
+        .await
+        .expect("creating a file for writing");
+
+    let (mut tx, rx) = wit_stream::new();
+    join! {
+        async {
+            file.write_via_stream(rx, 0).await.unwrap();
+        },
+        async {
+            loop {
+                // Wasmtime shouldn't buffer this much data by default on the
+                // host, something should have done a short write earlier.
+                assert!(len <= 128 << 20);
+                let (result, remaining) = tx.write(bytes(&mut pos, len)).await;
+                assert!(matches!(result, StreamResult::Complete(_)), "bad result {result:?}");
+                if remaining.remaining() == 0 {
+                    len = len.checked_mul(2).unwrap();
+                } else {
+                    pos -= remaining.remaining();
+                    break;
+                }
+            }
+            drop(tx);
+        },
+    };
+
+    let expected = bytes(&mut 0, pos);
+    let (rx, result) = file.read_via_stream(0);
+    let read_back = rx.collect().await;
+    result.await.unwrap();
+
+    assert_eq!(
+        read_back.len(),
+        expected.len(),
+        "wrong number of bytes read back"
+    );
+    assert!(
+        read_back == expected,
+        "contents differ after a chunked write"
+    );
+}
+
+fn main() {
+    unreachable!()
+}

crates/test-programs/src/bin/p3_file_write_chunked.rs

@@ -0,0 +1,86 @@
+use futures::join;
+use test_programs::p3::wasi::http::client;
+use test_programs::p3::wasi::http::types::{Headers, Method, Request, Response, Scheme};
+use test_programs::p3::{wit_future, wit_stream};
+use wit_bindgen::StreamResult;
+
+struct Component;
+
+test_programs::p3::export!(Component);
+
+fn bytes(offset: &mut usize, len: usize) -> Vec<u8> {
+    let mut buf = Vec::with_capacity(len);
+    for i in 0..len {
+        buf.push(((*offset + i) % 251) as u8);
+    }
+    *offset += len;
+    buf
+}
+
+fn addr() -> String {
+    test_programs::p3::wasi::cli::environment::get_environment()
+        .into_iter()
+        .find_map(|(k, v)| k.eq("HTTP_SERVER").then_some(v))
+        .unwrap()
+}
+
+impl test_programs::p3::exports::wasi::cli::run::Guest for Component {
+    async fn run() -> Result<(), ()> {
+        test_chunked_write().await;
+        Ok(())
+    }
+}
+
+async fn test_chunked_write() {
+    let headers = Headers::from_list(&[]).unwrap();
+    let (mut contents_tx, contents_rx) = wit_stream::new();
+    let (trailers_tx, trailers_rx) = wit_future::new(|| Ok(None));
+    let (request, transmit) = Request::new(headers, Some(contents_rx), trailers_rx, None);
+    configure(&request);
+
+    let (transmit, written, echoed) = join!(
+        async { transmit.await },
+        async {
+            let mut len = 16;
+            let mut pos = 0;
+            loop {
+                assert!(len <= 128 << 20);
+                let (result, remaining) = contents_tx.write(bytes(&mut pos, len)).await;
+                assert_eq!(result, StreamResult::Complete(len - remaining.remaining()));
+                if remaining.remaining() == 0 {
+                    len = len.checked_mul(2).unwrap();
+                } else {
+                    pos -= remaining.remaining();
+                    break;
+                }
+            }
+            drop(contents_tx);
+            _ = trailers_tx.write(Ok(None)).await;
+            pos
+        },
+        async { send_and_collect(request).await },
+    );
+    transmit.unwrap();
+    assert_eq!(echoed, bytes(&mut 0, written));
+}
+
+fn configure(request: &Request) {
+    request.set_method(&Method::Post).unwrap();
+    request.set_scheme(Some(&Scheme::Http)).unwrap();
+    request.set_authority(Some(&addr())).unwrap();
+    request.set_path_with_query(Some("/")).unwrap();
+}
+
+async fn send_and_collect(request: Request) -> Vec<u8> {
+    let response = client::send(request).await.unwrap();
+    assert_eq!(response.get_status_code(), 200);
+    let (_, result_rx) = wit_future::new(|| Ok(()));
+    let (body_rx, trailers_rx) = Response::consume_body(response, result_rx);
+    let body = body_rx.collect().await;
+    trailers_rx.await.unwrap();
+    body
+}
+
+fn main() {
+    unreachable!()
+}

crates/test-programs/src/bin/p3_http_outbound_request_chunk_size.rs

@@ -347,6 +347,13 @@ pub trait WasiHttpHooks: Send {
         })
     }
 
+    /// Maximum number of bytes the implementation will copy out of the guest in
+    /// a single write to an outgoing body's stream.
+    #[cfg(feature = "p3")]
+    fn p3_outgoing_body_chunk_size(&mut self) -> usize {
+        crate::p3::DEFAULT_OUTGOING_BODY_CHUNK_SIZE
+    }
+
     /// Optional hook to configure the error code for hyper errors.
     #[cfg(feature = "p3")]
     fn p3_error_from_hyper(&mut self, err: &hyper::Error) -> p3::ErrorCode {

crates/wasi-http/src/ctx.rs

@@ -177,6 +177,7 @@ struct LimitedGuestBodyConsumer {
     limit: u64,
     /// Number of bytes sent
     sent: u64,
+    max_chunk_size: usize,
     // `true` when the other side of `contents_tx` was unexpectedly closed
     closed: bool,
 }
@@ -217,7 +218,8 @@ impl<D> StreamConsumer<D> for LimitedGuestBodyConsumer {
         debug_assert!(!self.closed);
         let mut src = src.as_direct(store);
         let buf = src.remaining();
-        let n = buf.len();
+        let n = buf.len().min(self.max_chunk_size);
+        let buf = &buf[..n];
 
         // Perform `content-length` check early and precompute the next value
         let Ok(sent) = n.try_into() else {
@@ -260,7 +262,10 @@ impl<D> StreamConsumer<D> for LimitedGuestBodyConsumer {
 
 /// [StreamConsumer] implementation for bodies originating in the guest without `Content-Length`
 /// header set.
-struct UnlimitedGuestBodyConsumer(PollSender<Result<Bytes, ErrorCode>>);
+struct UnlimitedGuestBodyConsumer {
+    contents_tx: PollSender<Result<Bytes, ErrorCode>>,
+    max_chunk_size: usize,
+}
 
 impl<D> StreamConsumer<D> for UnlimitedGuestBodyConsumer {
     type Item = u8;
@@ -272,13 +277,13 @@ impl<D> StreamConsumer<D> for UnlimitedGuestBodyConsumer {
         src: Source<Self::Item>,
         finish: bool,
     ) -> Poll<wasmtime::Result<StreamResult>> {
-        match self.0.poll_reserve(cx) {
+        match self.contents_tx.poll_reserve(cx) {
             Poll::Ready(Ok(())) => {
                 let mut src = src.as_direct(store);
                 let buf = src.remaining();
-                let n = buf.len();
-                let buf = Bytes::copy_from_slice(buf);
-                match self.0.send_item(Ok(buf)) {
+                let n = buf.len().min(self.max_chunk_size);
+                let buf = Bytes::copy_from_slice(&buf[..n]);
+                match self.contents_tx.send_item(Ok(buf)) {
                     Ok(()) => {
                         src.mark_read(n);
                         Poll::Ready(Ok(StreamResult::Completed))
@@ -329,6 +334,11 @@ impl GuestBody {
             Ok(())
         })?;
 
+        let max_chunk_size = getter(store.as_context_mut().data_mut())
+            .hooks
+            .p3_outgoing_body_chunk_size()
+            .max(1);
+
         let contents_rx = if let Some(rx) = contents_rx {
             let (http_tx, http_rx) = mpsc::channel(1);
             let contents_tx = PollSender::new(http_tx);
@@ -348,12 +358,19 @@ impl GuestBody {
                         make_error,
                         limit,
                         sent: 0,
+                        max_chunk_size,
                         closed: false,
                     },
                 )?;
             } else {
                 _ = result_tx.send(Box::new(result_fut));
-                rx.pipe(store, UnlimitedGuestBodyConsumer(contents_tx))?;
+                rx.pipe(
+                    store,
+                    UnlimitedGuestBodyConsumer {
+                        contents_tx,
+                        max_chunk_size,
+                    },
+                )?;
             };
             Some(http_rx)
         } else {

crates/wasi-http/src/p3/body.rs

@@ -20,6 +20,11 @@ mod response;
 pub use request::Request;
 pub use response::Response;
 
+/// The default value configured for [`WasiHttpHooks::p3_outgoing_body_chunk_size`].
+///
+/// [`WasiHttpHooks::p3_outgoing_body_chunk_size`]: crate::WasiHttpHooks::p3_outgoing_body_chunk_size
+pub const DEFAULT_OUTGOING_BODY_CHUNK_SIZE: usize = 1024 * 1024;
+
 use crate::{FieldMapError, WasiHttp, WasiHttpView};
 use bindings::http::{client, types};
 use core::ops::Deref;

crates/wasi-http/src/p3/mod.rs

@@ -127,6 +127,7 @@ async fn run_cli(path: &str, server: &Server) -> wasmtime::Result<()> {
         Ctx {
             wasi: wasmtime_wasi::WasiCtx::builder()
                 .env("HTTP_SERVER", server.addr())
+                .inherit_stdio()
                 .build(),
             ..Ctx::new(oneshot::channel().0)
         },
@@ -870,3 +871,9 @@ async fn p3_http_empty_frames_interleaved() -> Result<()> {
     assert_eq!(collected_body, b"hello world".as_slice());
     Ok(())
 }
+
+#[test_log::test(tokio::test(flavor = "multi_thread"))]
+async fn p3_http_outbound_request_chunk_size() -> Result<()> {
+    let server = Server::http1(1)?;
+    run_cli(P3_HTTP_OUTBOUND_REQUEST_CHUNK_SIZE_COMPONENT, &server).await
+}

crates/wasi-http/tests/all/p3/mod.rs

@@ -463,7 +463,9 @@ impl<D> StreamConsumer<D> for WriteStreamConsumer {
         let me = &mut *self;
         let task = me.task.get_or_insert_with(|| {
             debug_assert!(me.buffer.is_empty());
-            me.buffer.extend_from_slice(src.remaining());
+            let remaining = src.remaining();
+            let n = remaining.len().min(DEFAULT_BUFFER_CAPACITY);
+            me.buffer.extend_from_slice(&remaining[..n]);
             let buf = mem::take(&mut me.buffer);
             let file = Arc::clone(me.file.as_file());
             let location = me.location;

crates/wasi/src/p3/filesystem/host.rs

@@ -185,6 +185,11 @@ async fn p3_file_write_blocking() -> wasmtime::Result<()> {
     run_allow_blocking_current_thread(P3_FILE_WRITE_COMPONENT, true).await
 }
 
+#[test_log::test(tokio::test(flavor = "multi_thread"))]
+async fn p3_file_write_chunked() -> wasmtime::Result<()> {
+    run(P3_FILE_WRITE_CHUNKED_COMPONENT).await
+}
+
 #[test_log::test(tokio::test(flavor = "multi_thread"))]
 async fn p3_file_truncation_readonly() -> wasmtime::Result<()> {
     run_with_readonly_testfile(P3_FILE_TRUNCATION_READONLY_COMPONENT).await