runtime-adapter: prove two clone-restore invariants from the host side

nvx-rs/README.md

@@ -523,6 +523,18 @@ and a process whose identity this user cannot check, such as another user's,
 makes `Process::open` fail rather than report an exit. Linux hosts without
 pidfds poll the reference instead, and cannot terminate the process.
 
+`Process::bytes_read` reports how many bytes the process has read through the
+operating system's read calls, whether the host's cache served them or not:
+`rchar` of `/proc/<pid>/io` on Linux, which it reads only while the process
+still runs with its reference's identity, and the `ReadTransferCount` of
+`GetProcessIoCounters` on Windows. `own_bytes_read` reports the calling
+process's. Neither counts what a process maps from a file, as OpenVMM maps the
+memory of a snapshot that it restores, nor, on Linux, what it reads through
+io_uring, as OpenVMM reads a sandbox's block devices. On Linux, the count also
+holds what the process's children read once it reaped them, as the kernel
+accounts them to their parent, while Windows counts the process's own reads
+alone.
+
 ### `nvx::host::supervisor`
 
 The supervisor of a sandbox's VMM, beneath the CLI, `Instance`, and external
@@ -927,6 +939,15 @@ The authenticated control session, which carries NVX's control protocol, `NVXC`,
 one frame per DATA record of a `HostConnection`; Linux and Windows primitives;
 and OpenVMM's outcome report.
 
+A `Transport` to an OpenVMM control endpoint reads and writes with deadlines,
+and tells, through `Transport::unread`, what the endpoint has not read yet of
+what it wrote, where the platform tells: on Linux, what `SIOCOUTQ` reports of
+its Unix socket, the kernel's charge for the buffers of each write until
+OpenVMM read each whole, which a read of part of a buffer leaves as it was, and
+which is zero once OpenVMM read everything. A named pipe tells its writer
+nothing of the kind, so on Windows, and for a transport that does not tell, it
+is `None`.
+
 ### `nvx::host::sandbox`
 
 The launch profile of `nvx-rs sandbox`, with native input validation:

nvx-rs/src/host/platform/linux.rs

@@ -188,6 +188,53 @@ impl ProcessHandle {
             result => super::unless_exited(result, || pidfd.wait_exit(Duration::ZERO)),
         }
     }
+
+    /// Returns how many bytes process `pid`, which the handle holds, and which started at
+    /// `start_time`, has read through read calls: the `rchar` of its `/proc/<pid>/io`.
+    ///
+    /// The file is opened before the process's identity is checked, so a process that still runs
+    /// with that start time held the ID when the file was opened, and the count is its own, with
+    /// or without a pidfd. Fails with [`io::ErrorKind::NotFound`] once the process has exited.
+    pub(crate) fn bytes_read(&self, pid: u32, start_time: u64) -> io::Result<u64> {
+        let path = format!("/proc/{pid}/io");
+        let exited = || io::Error::new(io::ErrorKind::NotFound, format!("process {pid} exited"));
+        let gone = |error: &io::Error| {
+            error.kind() == io::ErrorKind::NotFound
+                || error.raw_os_error() == Some(Errno::ESRCH as i32)
+        };
+        let mut file = match File::open(&path) {
+            Ok(file) => file,
+            Err(error) if gone(&error) => return Err(exited()),
+            Err(error) => return Err(error),
+        };
+        if process_start_time(pid)? != Some(start_time) {
+            return Err(exited());
+        }
+        let mut text = String::new();
+        match file.read_to_string(&mut text) {
+            Ok(_) => read_count(&path, &text),
+            Err(error) if gone(&error) => Err(exited()),
+            Err(error) => Err(error),
+        }
+    }
+}
+
+/// Returns how many bytes this process has read through read calls: the `rchar` of
+/// `/proc/self/io`.
+pub(crate) fn own_bytes_read() -> io::Result<u64> {
+    let path = "/proc/self/io";
+    read_count(path, &fs::read_to_string(path)?)
+}
+
+/// Returns the `rchar` count of the text `text` of the file at `path`, a process's `/proc` file
+/// `io`: the bytes that the process, and the children that it reaped, passed to read calls, from
+/// any file, whether the page cache served them or not, which counts no read through io_uring nor
+/// a file's mapped pages.
+fn read_count(path: &str, text: &str) -> io::Result<u64> {
+    text.lines()
+        .find_map(|line| line.strip_prefix("rchar:"))
+        .and_then(|count| count.trim().parse().ok())
+        .ok_or_else(|| io::Error::other(format!("{path} has no rchar count")))
 }
 
 /// Opens the process `pid` if it still runs with the start time `start_time`, and returns `None`
@@ -1097,6 +1144,10 @@ impl Transport for UnixTransport {
         }
         Ok(())
     }
+
+    fn unread(&self) -> io::Result<Option<usize>> {
+        sys::unread_by_peer(&self.stream).map(Some)
+    }
 }
 
 #[cfg(test)]
@@ -1212,4 +1263,48 @@ mod tests {
         drop(client);
         assert_eq!(receive_all(&mut listener), b"");
     }
+
+    #[test]
+    fn a_unix_transport_reports_what_its_peer_has_not_read() {
+        let (stream, mut peer) = UnixStream::pair().unwrap();
+        let mut transport = UnixTransport { stream };
+        assert_eq!(transport.unread().unwrap(), Some(0));
+        let data = b"an attachment";
+        transport.write_all(data, None).unwrap();
+        // The kernel charges the socket for the write's buffer, more than its bytes, until the
+        // peer read it whole: a read of a prefix leaves the charge as it was.
+        let queued = transport.unread().unwrap().unwrap();
+        assert!(queued >= data.len(), "{queued}");
+        let mut received = [0; 6];
+        peer.read_exact(&mut received).unwrap();
+        assert_eq!(transport.unread().unwrap(), Some(queued));
+        let mut rest = vec![0; data.len() - received.len()];
+        peer.read_exact(&mut rest).unwrap();
+        assert_eq!(transport.unread().unwrap(), Some(0));
+
+        // Writes of a byte each queue a buffer each, which the kernel charges the same for, and
+        // which the peer's read of any prefix frees.
+        let charges = data
+            .chunks(1)
+            .map(|byte| {
+                transport.write_all(byte, None).unwrap();
+                transport.unread().unwrap().unwrap()
+            })
+            .collect::<Vec<_>>();
+        let step = charges[0];
+        assert!(step > 0, "{charges:?}");
+        assert!(
+            charges
+                .iter()
+                .enumerate()
+                .all(|(index, charge)| *charge == (index + 1) * step),
+            "{charges:?}"
+        );
+        peer.read_exact(&mut received[..1]).unwrap();
+        let queued = charges[charges.len() - 1];
+        assert_eq!(transport.unread().unwrap(), Some(queued - step));
+        let mut rest = vec![0; data.len() - 1];
+        peer.read_exact(&mut rest).unwrap();
+        assert_eq!(transport.unread().unwrap(), Some(0));
+    }
 }

nvx-rs/src/host/platform/mod.rs

@@ -54,6 +54,19 @@ pub trait Transport: Send {
     /// [`io::ErrorKind::TimedOut`] error if it has not by then, however much of it the endpoint
     /// reads meanwhile.
     fn write_all(&mut self, data: &[u8], timeout: Option<Duration>) -> io::Result<()>;
+
+    /// Returns what the platform tells of the data that this end wrote, and that the endpoint has
+    /// not read yet, or `None` where it tells nothing.
+    ///
+    /// On Linux, it is what `SIOCOUTQ` reports of the Unix socket: the memory that the kernel
+    /// charges the socket for the buffers that hold that data, each write's buffers of their own,
+    /// until the endpoint read each buffer whole. The charge exceeds the data's bytes, and a read
+    /// of part of a buffer leaves it as it was, so only data written a byte at a time shows the
+    /// read of each byte. It is zero once the endpoint read everything. A named pipe tells its
+    /// writer nothing of the kind, so on Windows, and by default, it is `None`.
+    fn unread(&self) -> io::Result<Option<usize>> {
+        Ok(None)
+    }
 }
 
 /// Returns the time left until `deadline`, `None` without one, or an [`io::ErrorKind::TimedOut`]

nvx-rs/src/host/platform/other.rs

@@ -56,6 +56,16 @@ impl ProcessHandle {
     pub(crate) fn kill(&self, _pid: u32) -> io::Result<()> {
         match *self {}
     }
+
+    /// Never called: no handle exists.
+    pub(crate) fn bytes_read(&self, _pid: u32, _start_time: u64) -> io::Result<u64> {
+        match *self {}
+    }
+}
+
+/// Fails: this host has no read counters that NVX reads.
+pub(crate) fn own_bytes_read() -> io::Result<u64> {
+    Err(unsupported())
 }
 
 /// Reports every process as gone, as [`process_start_time`] does.

nvx-rs/src/host/platform/windows.rs

@@ -66,11 +66,12 @@ use windows_sys::Win32::System::Threading::GetExitCodeProcess;
 use windows_sys::Win32::System::Threading::{
     CREATE_BREAKAWAY_FROM_JOB, CREATE_NEW_PROCESS_GROUP, CREATE_NO_WINDOW,
     CREATE_UNICODE_ENVIRONMENT, CreateEventW, CreateProcessW, DeleteProcThreadAttributeList,
-    EXTENDED_STARTUPINFO_PRESENT, GetCurrentProcess, GetProcessTimes, INFINITE,
-    InitializeProcThreadAttributeList, LPPROC_THREAD_ATTRIBUTE_LIST, OpenProcess, OpenProcessToken,
-    PROC_THREAD_ATTRIBUTE_HANDLE_LIST, PROC_THREAD_ATTRIBUTE_JOB_LIST, PROCESS_CREATION_FLAGS,
-    PROCESS_INFORMATION, PROCESS_QUERY_LIMITED_INFORMATION, PROCESS_SYNCHRONIZE, PROCESS_TERMINATE,
-    ResetEvent, STARTF_USESTDHANDLES, STARTUPINFOEXW, TerminateProcess, UpdateProcThreadAttribute,
+    EXTENDED_STARTUPINFO_PRESENT, GetCurrentProcess, GetProcessIoCounters, GetProcessTimes,
+    INFINITE, IO_COUNTERS, InitializeProcThreadAttributeList, LPPROC_THREAD_ATTRIBUTE_LIST,
+    OpenProcess, OpenProcessToken, PROC_THREAD_ATTRIBUTE_HANDLE_LIST,
+    PROC_THREAD_ATTRIBUTE_JOB_LIST, PROCESS_CREATION_FLAGS, PROCESS_INFORMATION,
+    PROCESS_QUERY_LIMITED_INFORMATION, PROCESS_SYNCHRONIZE, PROCESS_TERMINATE, ResetEvent,
+    STARTF_USESTDHANDLES, STARTUPINFOEXW, TerminateProcess, UpdateProcThreadAttribute,
     WaitForSingleObject,
 };
 use windows_sys::Win32::System::WindowsProgramming::QueryInterruptTimePrecise;
@@ -150,6 +151,27 @@ fn has_exited(process: &OwnedHandle) -> io::Result<bool> {
     }
 }
 
+/// Returns how many bytes the process of `process`, a handle with query access, has read through
+/// read calls: the `ReadTransferCount` of `GetProcessIoCounters`, which counts every `ReadFile`,
+/// whatever it reads, and whether the cache served it or not.
+fn read_transfer(process: HANDLE) -> io::Result<u64> {
+    let mut counters = IO_COUNTERS::default();
+    // SAFETY: the handle has query access, and the output pointer is valid.
+    if unsafe { GetProcessIoCounters(process, &mut counters) } != 0 {
+        Ok(counters.ReadTransferCount)
+    } else {
+        Err(io::Error::last_os_error())
+    }
+}
+
+/// Returns how many bytes this process has read through read calls, as [`read_transfer`] counts
+/// them.
+pub(crate) fn own_bytes_read() -> io::Result<u64> {
+    // SAFETY: GetCurrentProcess has no preconditions, and returns a pseudo-handle with every
+    // access, which needs no closing.
+    read_transfer(unsafe { GetCurrentProcess() })
+}
+
 /// Returns the outcome of a termination that Windows refused with `error` through a handle with
 /// terminate access: success if the process is exiting or has exited, and `error` otherwise.
 ///
@@ -273,6 +295,13 @@ impl ProcessHandle {
         )
     }
 
+    /// Returns how many bytes the process has read through read calls, as [`read_transfer`]
+    /// counts them: once the process has exited, those that it read until then, which the handle
+    /// keeps.
+    pub(crate) fn bytes_read(&self, _pid: u32, _start_time: u64) -> io::Result<u64> {
+        read_transfer(raw(&self.process))
+    }
+
     /// Closes the handle of process `pid` that has the value of this process's handle `ours`,
     /// which the process inherited from this one: a handle keeps its value in a child that
     /// inherits it, and a child that never uses it, as a VMM never uses the guards of the checked

nvx-rs/src/host/process.rs

@@ -15,6 +15,10 @@
 //! Linux 5.3 or in sandboxes that refuse them, check the reference at a short interval instead,
 //! until the deadline of the wait, and cannot terminate a process that they cannot hold.
 //!
+//! [`Process::bytes_read`] reports how many bytes the process has read through the operating
+//! system's read calls, and [`own_bytes_read`] how many the calling process has, so that a caller
+//! can tell how much of its files a VMM, or the caller itself, read.
+//!
 //! The [`supervisor`](super::supervisor) launches OpenVMM, and its
 //! [`Launched::process`](super::supervisor::Launched::process) holds the child that it started
 //! through a handle of its own.
@@ -240,6 +244,27 @@ impl Process {
         self.handle.kill(self.reference.pid)
     }
 
+    /// Returns how many bytes the process has read through the operating system's read calls,
+    /// from files, pipes, sockets, and devices alike, whether the host's cache served them or not:
+    /// the `rchar` of `/proc/<pid>/io` on Linux, and the `ReadTransferCount` of
+    /// `GetProcessIoCounters` on Windows. Neither counts what the process reads otherwise: the
+    /// pages that it maps from a file, as OpenVMM maps the memory of a snapshot that it restores,
+    /// nor, on Linux, what it reads through io_uring, as OpenVMM reads a sandbox's block devices.
+    /// On Linux, the count also holds what the process's children read, once the process reaped
+    /// them, as the kernel accounts them to their parent, while Windows counts the process's own
+    /// reads alone.
+    ///
+    /// On Linux, the count is read from the process's `/proc` file, which is opened before the
+    /// process's identity is checked, so that the count is the process's own, and the read fails
+    /// with [`io::ErrorKind::NotFound`] once the process has exited. On Windows, the handle keeps
+    /// the count of a process that exited. Fails with [`io::ErrorKind::PermissionDenied`] when
+    /// this user may not read the process's counters, and with [`io::ErrorKind::Unsupported`] on
+    /// hosts other than Linux and Windows.
+    pub fn bytes_read(&self) -> io::Result<u64> {
+        self.handle
+            .bytes_read(self.reference.pid, self.reference.start_time)
+    }
+
     /// Closes the process's handle that has the value of this process's handle `ours`, which the
     /// process inherited from this one, and never uses, as a VMM never uses the guards of the
     /// checked files that it inherits on Windows.
@@ -266,6 +291,15 @@ impl fmt::Debug for Process {
     }
 }
 
+/// Returns how many bytes the calling process has read through the operating system's read
+/// calls, as [`Process::bytes_read`] counts them: on Linux, with what the children that it reaped
+/// read.
+///
+/// Fails with [`io::ErrorKind::Unsupported`] on hosts other than Linux and Windows.
+pub fn own_bytes_read() -> io::Result<u64> {
+    platform::own_bytes_read()
+}
+
 /// Checks `reference` against the boot that `current_boot` returns, and during the same boot,
 /// opens the process with `open`, as [`Process::open`] does.
 fn open_reference<H>(
@@ -546,6 +580,85 @@ mod reference_tests {
                 .unwrap()
         );
     }
+
+    /// Returns a command that reads the first line of the file at `path`, prints `read`, then
+    /// waits for a line of its standard input, which reaches its end once the test closes it, and
+    /// exits. The shell reads the line itself, through read calls of its own.
+    #[cfg(windows)]
+    fn reader(path: &std::path::Path) -> std::process::Command {
+        use std::os::windows::process::CommandExt;
+
+        let mut command = std::process::Command::new("cmd");
+        // `cmd` takes the rest of its command line as it is, the quotes around the path included.
+        command.args(["/d", "/q", "/c"]).raw_arg(format!(
+            "set /p line=<\"{}\" & echo read & set /p wait=",
+            path.display()
+        ));
+        command
+            .stdin(std::process::Stdio::piped())
+            .stdout(std::process::Stdio::piped());
+        command
+    }
+
+    /// Returns a command that reads the first line of the file at `path`, prints `read`, then
+    /// waits for a line of its standard input, which reaches its end once the test closes it, and
+    /// exits. The shell reads the line itself, through read calls of its own.
+    #[cfg(target_os = "linux")]
+    fn reader(path: &std::path::Path) -> std::process::Command {
+        let mut command = std::process::Command::new("sh");
+        command
+            .args(["-c", "read -r line < \"$0\"; echo read; read -r wait"])
+            .arg(path)
+            .stdin(std::process::Stdio::piped())
+            .stdout(std::process::Stdio::piped());
+        command
+    }
+
+    #[cfg(any(target_os = "linux", windows))]
+    #[test]
+    fn processes_count_the_bytes_that_their_read_calls_read() {
+        use std::io::{BufRead, BufReader, Write};
+
+        // A line short enough for `set /p`, which reads at most 1023 characters of one.
+        const LINE: usize = 512;
+        let path = std::env::temp_dir().join(format!("nvx-bytes-read-{}", std::process::id()));
+        let mut contents = vec![b'x'; LINE];
+        contents.push(b'\n');
+        std::fs::write(&path, &contents).unwrap();
+
+        // This process counts what it reads, and what other tests' threads read meanwhile.
+        let before = own_bytes_read().unwrap();
+        let read = std::fs::read(&path).unwrap();
+        assert!(own_bytes_read().unwrap() >= before + read.len() as u64);
+
+        // Another process counts what it read once it says so, while it runs.
+        let mut child = platform::inherit::spawn_command(&mut reader(&path)).unwrap();
+        let mut said = String::new();
+        let stdout = child.stdout.take().unwrap();
+        BufReader::new(stdout).read_line(&mut said).unwrap();
+        let reference = ProcessRef {
+            pid: child.id(),
+            start_time: platform::process_start_time(child.id()).unwrap().unwrap(),
+            boot: BootId::current().unwrap(),
+        };
+        let process = Process::open(&reference).unwrap().unwrap();
+        let counted = process.bytes_read();
+        let mut stdin = child.stdin.take().unwrap();
+        stdin.write_all(b"x\n").unwrap();
+        drop(stdin);
+        child.wait().unwrap();
+        std::fs::remove_file(&path).unwrap();
+        assert_eq!(said.trim_end(), "read");
+        assert!(counted.unwrap() >= LINE as u64);
+        // Once the process exited, Linux keeps no count of it, and Windows keeps its count in the
+        // handle.
+        let exited = process.bytes_read();
+        if cfg!(windows) {
+            assert!(exited.unwrap() >= LINE as u64);
+        } else {
+            assert_eq!(exited.unwrap_err().kind(), io::ErrorKind::NotFound);
+        }
+    }
 }
 
 #[cfg(all(test, target_os = "linux"))]

nvx-rs/src/host/sys/mod.rs

@@ -35,7 +35,7 @@ pub(crate) use process::{
 pub(crate) use process::{clear_close_on_exec, mark_listed_descriptors_on_spawn};
 #[cfg(target_os = "linux")]
 pub(crate) use socket::{
-    connect_nonblocking, peer_pid, send_nonblocking, wait_readable, wait_writable,
+    connect_nonblocking, peer_pid, send_nonblocking, unread_by_peer, wait_readable, wait_writable,
 };
 
 /// Returns the effective user and group IDs of this process.