Fix stdin over-read on seekable inputs

crates/wasi/src/cli/worker_thread_stdin.rs

@@ -25,7 +25,6 @@
 
 use crate::cli::{IsTerminal, StdinStream};
 use bytes::{Bytes, BytesMut};
-use std::io::Read;
 use std::mem;
 use std::pin::Pin;
 use std::sync::{Condvar, Mutex, OnceLock};
@@ -116,7 +115,7 @@ fn create() -> GlobalStdin {
             drop(lock);
 
             let mut bytes = BytesMut::zeroed(size_hint);
-            let (new_state, done) = match std::io::stdin().read(&mut bytes) {
+            let (new_state, done) = match read_stdin(&mut bytes) {
                 Ok(0) => (StdinState::Closed, true),
                 Ok(nbytes) => {
                     bytes.truncate(nbytes);
@@ -143,6 +142,44 @@ fn create() -> GlobalStdin {
     GlobalStdin::default()
 }
 
+// Bypass `std::io::Stdin`'s process-global buffer so that a guest request
+// cannot advance a seekable input past the requested number of bytes. Keep
+// its lock held to serialize reads with other users of stdin in this process.
+fn read_stdin(bytes: &mut [u8]) -> std::io::Result<usize> {
+    #[cfg(unix)]
+    {
+        use std::os::fd::AsFd;
+        let stdin = std::io::stdin();
+        let stdin = stdin.lock();
+        rustix::io::read(stdin.as_fd(), bytes).map_err(Into::into)
+    }
+
+    #[cfg(windows)]
+    {
+        use std::io::Read as _;
+        use std::os::windows::io::{AsRawHandle, FromRawHandle};
+
+        let stdin = std::io::stdin();
+        let mut stdin = stdin.lock();
+        if std::io::IsTerminal::is_terminal(&stdin) {
+            return stdin.read(bytes);
+        }
+
+        // SAFETY: `stdin` keeps the borrowed process handle valid for this
+        // read, and `ManuallyDrop` prevents `File` from closing the handle.
+        let mut file = std::mem::ManuallyDrop::new(unsafe {
+            std::fs::File::from_raw_handle(stdin.as_raw_handle())
+        });
+        file.read(bytes)
+    }
+
+    #[cfg(not(any(unix, windows)))]
+    {
+        use std::io::Read as _;
+        std::io::stdin().read(bytes)
+    }
+}
+
 struct WasiStdin;
 
 #[async_trait::async_trait]

crates/wasi/tests/process_stdin.rs

@@ -1,4 +1,4 @@
-use std::io::{BufRead, Write};
+use std::io::{BufRead, Seek, Write};
 use std::process::Command;
 use wasmtime_wasi::cli::StdinStream;
 use wasmtime_wasi::p2::Pollable;
@@ -16,11 +16,30 @@ fn main() {
         return;
     }
 
-    match std::env::var(VAR_NAME) {
+    match std::env::var(VAR_NAME).as_deref() {
+        Ok("read-once") => child_read_once(),
         Ok(_) => child_process(),
         Err(_) => parent_process(),
     }
 
+    fn child_read_once() {
+        tokio::runtime::Builder::new_multi_thread()
+            .enable_all()
+            .build()
+            .unwrap()
+            .block_on(async {
+                let mut stdin = wasmtime_wasi::cli::stdin().p2_stream();
+                loop {
+                    let bytes = stdin.read(1).unwrap();
+                    if !bytes.is_empty() {
+                        assert_eq!(&bytes[..], b"a");
+                        break;
+                    }
+                    stdin.ready().await;
+                }
+            });
+    }
+
     fn child_process() {
         let mut result_write = std::io::stderr();
         let mut child_running = true;
@@ -92,6 +111,18 @@ fn main() {
 
 fn parent_process() {
     let me = std::env::current_exe().unwrap();
+
+    let mut input = tempfile::tempfile().unwrap();
+    input.write_all(b"abcde").unwrap();
+    input.rewind().unwrap();
+    let status = Command::new(&me)
+        .env(VAR_NAME, "read-once")
+        .stdin(input.try_clone().unwrap())
+        .status()
+        .unwrap();
+    assert!(status.success());
+    assert_eq!(input.stream_position().unwrap(), 1);
+
     let mut cmd = Command::new(me);
     cmd.env(VAR_NAME, "1");
     cmd.stdin(std::process::Stdio::piped());