`wasi-http` not handling 0-length response frames

[karthik-phl]

Thanks for filing a bug report! Please fill out the TODOs below.

Note: if you want to report a security issue, please read our security policy!

Test Case

Use Reqwest for outbound HTTP calls that are sent by the wasi-http host

Steps to Reproduce

• We've created an implementation of the wasi-http host that uses Reqwest 0.13 for outbound calls. When using wit-bindgen's async-stream support to process the response of an outbound call through the host, we found that wasmtime encounters a situation that causes the guest module to crash.

• Upon further investigation, we found that Reqwest sends a 0-length frame when the end of stream is reached, which is not handled by the poll_produce method of StreamProducer. I think wasmtime should handle this case explicitly and prevent the guest from crashing, and I'm happy to create a PR with this change.

• The following test that can be added to wasi-http/tests/all/p3/mod.rs to simulate the condition that's triggered by the use of a library like Reqwest. Running the test with cargo test -p wasmtime-wasi-http --features p3 --test all p3_http_empty_frame_at_end_of_stream -- --nocapture should indicate the error.

use std::pin::Pin;
use std::task::{Context, Poll};

// Custom body wrapper that sends an empty frame at EOS while reporting is_end_stream() = true
struct BodyWithEmptyFrameAtEos {
    inner: http_body_util::StreamBody<
        futures::channel::mpsc::Receiver<Result<http_body::Frame<Bytes>, ErrorCode>>,
    >,
    sent_empty: bool,
    at_eos: bool,
}

impl http_body::Body for BodyWithEmptyFrameAtEos {
    type Data = Bytes;
    type Error = ErrorCode;

    fn poll_frame(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Result<http_body::Frame<Self::Data>, Self::Error>>> {
        // First, poll the underlying body
        let this = &mut *self;
        match Pin::new(&mut this.inner).poll_frame(cx) {
            Poll::Ready(None) if !this.sent_empty => {
                // When the underlying body ends, send an empty frame
                // This simulates HTTP implementations that send empty frames at EOS
                this.sent_empty = true;
                this.at_eos = true;
                Poll::Ready(Some(Ok(http_body::Frame::data(Bytes::new()))))
            }
            other => other,
        }
    }

    fn is_end_stream(&self) -> bool {
        // Report end of stream once we've reached it
        // This ensures is_end_stream() = true when we send the empty frame
        self.at_eos
    }
}

#[test_log::test(tokio::test(flavor = "multi_thread"))]
async fn p3_http_empty_frame_at_end_of_stream() -> Result<()> {
    _ = env_logger::try_init();

    // This test verifies the fix which handles the case where a zero-length frame is
    // received when is_end_stream() is true. Without the fix, the StreamProducer would
    // crash when the WASM guest tries to read such a frame.

    let body = b"test";
    let raw_body = Bytes::copy_from_slice(body);

    let (mut body_tx, body_rx) = futures::channel::mpsc::channel::<Result<_, ErrorCode>>(1);

    let wrapped_body = BodyWithEmptyFrameAtEos {
        inner: http_body_util::StreamBody::new(body_rx),
        sent_empty: false,
        at_eos: false,
    };

    let request = http::Request::builder()
        .uri("http://localhost/")
        .method(http::Method::GET);

    // Use the echo component which actually reads from the stream
    let response = futures::join!(
        run_http(
            P3_HTTP_ECHO_COMPONENT,
            request.body(wrapped_body)?,
            oneshot::channel().0
        ),
        async {
            body_tx
                .send(Ok(http_body::Frame::data(raw_body)))
                .await
                .unwrap();
            drop(body_tx);
        }
    )
    .0?
    .unwrap();

    assert_eq!(response.status().as_u16(), 200);

    // Verify the body was echoed correctly (empty frames should be filtered out by the fix)
    let (_, collected_body) = response.into_parts();
    let collected_body = collected_body.to_bytes();
    assert_eq!(collected_body, body.as_slice());
    Ok(())
}

Thanks for looking into this!

Expected Results

The WASM guest should be able to collect the stream results and continue processing

Actual Results

The WASM guest crashes abruptly

Versions and Environment

Wasmtime version or commit: 41.0.x