From 0a479b5800f015f46302e137b97925364a61c4de Mon Sep 17 00:00:00 2001 From: Surma Date: Fri, 23 Oct 2026 00:08:02 +0100 Subject: [PATCH] sausage-vfs: layers (host, memory, generated, mounts), process guards, wasi:filesystem adapter Checkpoint. More tests and the review findings follow. --- vfs/Cargo.toml | 11 + vfs/src/adapter.rs | 470 +++++++++++++++++++++++++++++ vfs/src/generated.rs | 132 ++++++++ vfs/src/host.rs | 190 ++++++++++++ vfs/src/lib.rs | 93 +++++- vfs/src/memory.rs | 257 ++++++++++++++++ vfs/src/mounts.rs | 97 ++++++ vfs/src/process.rs | 51 ++++ vfs/src/readonly.rs | 30 ++ vfs/src/streams.rs | 134 ++++++++ vfs/tests/guests/files/Cargo.toml | 8 + vfs/tests/guests/files/src/main.rs | 66 ++++ vfs/tests/layers.rs | 233 ++++++++++++++ vfs/tests/streams.rs | 67 ++++ vfs/tests/wasi.rs | 94 ++++++ 15 files changed, 1932 insertions(+), 1 deletion(-) create mode 100644 vfs/src/adapter.rs create mode 100644 vfs/src/generated.rs create mode 100644 vfs/src/host.rs create mode 100644 vfs/src/memory.rs create mode 100644 vfs/src/mounts.rs create mode 100644 vfs/src/process.rs create mode 100644 vfs/src/readonly.rs create mode 100644 vfs/src/streams.rs create mode 100644 vfs/tests/guests/files/Cargo.toml create mode 100644 vfs/tests/guests/files/src/main.rs create mode 100644 vfs/tests/layers.rs create mode 100644 vfs/tests/streams.rs create mode 100644 vfs/tests/wasi.rs diff --git a/vfs/Cargo.toml b/vfs/Cargo.toml index 6704733..6da482b 100644 --- a/vfs/Cargo.toml +++ b/vfs/Cargo.toml @@ -5,3 +5,14 @@ edition = "2024" description = "A composable virtual filesystem for WASI guests in Wasmtime." [dependencies] +async-trait = "0.1" +cap-std = "4" +wasmtime = "49" +wasmtime-wasi = "49" +bytes = "1" +tokio = { version = "1", features = ["rt-multi-thread", "sync", "macros"] } + +[dev-dependencies] +proptest = "1" +tempfile = "3" +rusqlite = { version = "0.38", features = ["bundled"] } diff --git a/vfs/src/adapter.rs b/vfs/src/adapter.rs new file mode 100644 index 0000000..98cb8b1 --- /dev/null +++ b/vfs/src/adapter.rs @@ -0,0 +1,470 @@ +//! Wasmtime 49 filesystem interfaces backed exclusively by the virtual mount namespace. +use crate::streams::{Reader, Writer}; +use crate::*; +use wasmtime::component::{HasData, Linker, Resource, ResourceTable}; +use wasmtime_wasi::p2::{ + DynInputStream, DynOutputStream, FsError, FsResult, + bindings::filesystem::{ + preopens, + types::{ + self, DescriptorFlags, DescriptorType, DirectoryEntry, ErrorCode, HostDescriptor, + HostDirectoryEntryStream, OpenFlags, + }, + }, +}; + +/// Store-owned filesystem namespace. +pub struct Adapter { + pub mounts: Arc, +} +impl Adapter { + pub fn new(mounts: Mounts) -> Self { + Self { + mounts: Arc::new(mounts), + } + } +} +/// The filesystem and shared WASI resource table borrowed from store state. +pub struct View<'a> { + pub adapter: &'a mut Adapter, + pub table: &'a mut ResourceTable, +} +pub struct Vfs; +impl HasData for Vfs { + type Data<'a> = View<'a>; +} +/// Replace filesystem imports after adding the standard asynchronous WASI linker interfaces. +pub fn add_to_linker( + linker: &mut Linker, + view: fn(&mut T) -> View<'_>, +) -> wasmtime::Result<()> { + linker.allow_shadowing(true); + types::add_to_linker::(linker, view)?; + preopens::add_to_linker::(linker, view)?; + Ok(()) +} +#[derive(Clone)] +struct Descriptor { + path: String, + file: Arc, + flags: DescriptorFlags, +} +struct Entries(std::vec::IntoIter); +fn code(e: &io::Error) -> ErrorCode { + use io::ErrorKind::*; + match e.kind() { + NotADirectory => ErrorCode::NotDirectory, + IsADirectory => ErrorCode::IsDirectory, + DirectoryNotEmpty => ErrorCode::NotEmpty, + ReadOnlyFilesystem => ErrorCode::ReadOnly, + CrossesDevices => ErrorCode::CrossDevice, + Unsupported => ErrorCode::Unsupported, + _ => ErrorCode::from(e), + } +} +fn errno(e: io::Error) -> FsError { + code(&e).into() +} +fn stat(s: Stat) -> types::DescriptorStat { + fn timestamp(t: SystemTime) -> Option { + let d = t.duration_since(SystemTime::UNIX_EPOCH).ok()?; + Some(types::Datetime { + seconds: d.as_secs(), + nanoseconds: d.subsec_nanos(), + }) + } + types::DescriptorStat { + type_: if s.directory { + types::DescriptorType::Directory + } else { + types::DescriptorType::RegularFile + }, + link_count: 1, + size: s.size, + data_access_timestamp: s.accessed.and_then(timestamp), + data_modification_timestamp: s.modified.and_then(timestamp), + status_change_timestamp: None, + } +} +fn time(t: types::NewTimestamp) -> Option { + match t { + types::NewTimestamp::NoChange => None, + types::NewTimestamp::Now => Some(SystemTime::now()), + types::NewTimestamp::Timestamp(t) => { + Some(SystemTime::UNIX_EPOCH + std::time::Duration::new(t.seconds, t.nanoseconds)) + } + } +} +fn hash(path: &str) -> types::MetadataHashValue { + use std::hash::{Hash, Hasher}; + let mut h = std::collections::hash_map::DefaultHasher::new(); + path.hash(&mut h); + types::MetadataHashValue { + lower: h.finish(), + upper: 0, + } +} +impl View<'_> { + fn fd(&self, fd: &Resource) -> FsResult { + Ok(self + .table + .get(&Resource::::new_borrow(fd.rep()))? + .clone()) + } + fn path(&self, fd: &Resource, path: &str) -> FsResult { + Ok(resolve(&self.fd(fd)?.path, path)) + } + fn push(&mut self, descriptor: Descriptor) -> FsResult> { + Ok(Resource::new_own(self.table.push(descriptor)?.rep())) + } +} +impl preopens::Host for View<'_> { + fn get_directories(&mut self) -> wasmtime::Result, String)>> { + // The root descriptor needs no IO: directory operations use its virtual path. + let file: Arc = Arc::new(Root); + let fd = self.push(Descriptor { + path: "/".into(), + file, + flags: DescriptorFlags::READ | DescriptorFlags::MUTATE_DIRECTORY, + })?; + Ok(vec![(fd, "/".into())]) + } +} +struct Root; +#[async_trait] +impl File for Root { + async fn stat(&self) -> Result { + Ok(Stat { + directory: true, + size: 0, + accessed: None, + modified: None, + }) + } + async fn read(&self, _: u64, _: usize) -> Result> { + Err(error(io::ErrorKind::IsADirectory)) + } + async fn write(&self, _: u64, _: Vec) -> Result { + Err(error(io::ErrorKind::IsADirectory)) + } + async fn append(&self, _: Vec) -> Result { + Err(error(io::ErrorKind::IsADirectory)) + } + async fn set_size(&self, _: u64) -> Result<()> { + Err(error(io::ErrorKind::IsADirectory)) + } + async fn sync(&self) -> Result<()> { + Ok(()) + } + async fn set_times(&self, _: Option, _: Option) -> Result<()> { + Err(error(io::ErrorKind::Unsupported)) + } +} +impl types::Host for View<'_> { + fn convert_error_code(&mut self, err: FsError) -> wasmtime::Result { + err.downcast() + } + fn filesystem_error_code( + &mut self, + err: Resource, + ) -> wasmtime::Result> { + Ok(self.table.get(&err)?.downcast_ref::().map(code)) + } +} + +impl HostDescriptor for View<'_> { + async fn advise( + &mut self, + fd: Resource, + offset: types::Filesize, + len: types::Filesize, + advice: types::Advice, + ) -> FsResult<()> { + let _ = (fd, offset, len, advice); + Ok(()) + } + async fn sync_data(&mut self, fd: Resource) -> FsResult<()> { + self.fd(&fd)?.file.sync().await.map_err(errno) + } + async fn get_flags(&mut self, fd: Resource) -> FsResult { + Ok(self.fd(&fd)?.flags) + } + async fn get_type(&mut self, fd: Resource) -> FsResult { + Ok(stat(self.fd(&fd)?.file.stat().await.map_err(errno)?).type_) + } + async fn set_size( + &mut self, + fd: Resource, + size: types::Filesize, + ) -> FsResult<()> { + self.fd(&fd)?.file.set_size(size).await.map_err(errno) + } + async fn set_times( + &mut self, + fd: Resource, + atim: types::NewTimestamp, + mtim: types::NewTimestamp, + ) -> FsResult<()> { + self.fd(&fd)? + .file + .set_times(time(atim), time(mtim)) + .await + .map_err(errno) + } + async fn read( + &mut self, + fd: Resource, + len: types::Filesize, + offset: types::Filesize, + ) -> FsResult<(Vec, bool)> { + let bytes = self + .fd(&fd)? + .file + .read(offset, len as usize) + .await + .map_err(errno)?; + let eof = bytes.is_empty(); + Ok((bytes, eof)) + } + async fn write( + &mut self, + fd: Resource, + buf: Vec, + offset: types::Filesize, + ) -> FsResult { + self.fd(&fd)? + .file + .write(offset, buf) + .await + .map(|n| n as u64) + .map_err(errno) + } + async fn read_directory( + &mut self, + fd: Resource, + ) -> FsResult> { + let d = self.fd(&fd)?; + let entries = self + .adapter + .mounts + .list(&d.path) + .await + .map_err(errno)? + .into_iter() + .map(|e| DirectoryEntry { + name: e.name, + type_: if e.directory { + types::DescriptorType::Directory + } else { + types::DescriptorType::RegularFile + }, + }) + .collect::>(); + Ok(Resource::new_own( + self.table.push(Entries(entries.into_iter()))?.rep(), + )) + } + async fn sync(&mut self, fd: Resource) -> FsResult<()> { + self.fd(&fd)?.file.sync().await.map_err(errno) + } + async fn create_directory_at( + &mut self, + fd: Resource, + path: String, + ) -> FsResult<()> { + let p = self.path(&fd, &path)?; + self.adapter.mounts.mkdir(&p).await.map_err(errno) + } + async fn stat(&mut self, fd: Resource) -> FsResult { + Ok(stat(self.fd(&fd)?.file.stat().await.map_err(errno)?)) + } + async fn stat_at( + &mut self, + fd: Resource, + path_flags: types::PathFlags, + path: String, + ) -> FsResult { + let _ = path_flags; + let p = self.path(&fd, &path)?; + Ok(stat(self.adapter.mounts.stat(&p).await.map_err(errno)?)) + } + async fn set_times_at( + &mut self, + fd: Resource, + path_flags: types::PathFlags, + path: String, + atim: types::NewTimestamp, + mtim: types::NewTimestamp, + ) -> FsResult<()> { + let _ = path_flags; + let p = self.path(&fd, &path)?; + self.adapter + .mounts + .open( + &p, + Open { + write: true, + ..Open::default() + }, + ) + .await + .map_err(errno)? + .set_times(time(atim), time(mtim)) + .await + .map_err(errno) + } + async fn link_at( + &mut self, + fd: Resource, + old_path_flags: types::PathFlags, + old_path: String, + new_descriptor: Resource, + new_path: String, + ) -> FsResult<()> { + let _ = (fd, old_path_flags, old_path, new_descriptor, new_path); + Err(ErrorCode::Unsupported.into()) + } + async fn open_at( + &mut self, + fd: Resource, + path_flags: types::PathFlags, + path: String, + oflags: OpenFlags, + flags: DescriptorFlags, + ) -> FsResult> { + let _ = path_flags; + let path = self.path(&fd, &path)?; + let file = self + .adapter + .mounts + .open( + &path, + Open { + write: flags.contains(DescriptorFlags::WRITE), + create: oflags.contains(OpenFlags::CREATE), + exclusive: oflags.contains(OpenFlags::EXCLUSIVE), + truncate: oflags.contains(OpenFlags::TRUNCATE), + directory: oflags.contains(OpenFlags::DIRECTORY), + }, + ) + .await + .map_err(errno)?; + self.push(Descriptor { path, file, flags }) + } + fn drop(&mut self, fd: Resource) -> wasmtime::Result<()> { + self.table + .delete(Resource::::new_own(fd.rep()))?; + Ok(()) + } + async fn readlink_at( + &mut self, + fd: Resource, + path: String, + ) -> FsResult { + let _ = (fd, path); + Err(ErrorCode::Unsupported.into()) + } + async fn remove_directory_at( + &mut self, + fd: Resource, + path: String, + ) -> FsResult<()> { + let p = self.path(&fd, &path)?; + self.adapter.mounts.remove_dir(&p).await.map_err(errno) + } + async fn rename_at( + &mut self, + fd: Resource, + old_path: String, + new_fd: Resource, + new_path: String, + ) -> FsResult<()> { + let a = self.path(&fd, &old_path)?; + let b = self.path(&new_fd, &new_path)?; + self.adapter.mounts.rename(&a, &b).await.map_err(errno) + } + async fn symlink_at( + &mut self, + fd: Resource, + src_path: String, + dest_path: String, + ) -> FsResult<()> { + let _ = (fd, src_path, dest_path); + Err(ErrorCode::Unsupported.into()) + } + async fn unlink_file_at( + &mut self, + fd: Resource, + path: String, + ) -> FsResult<()> { + let p = self.path(&fd, &path)?; + self.adapter.mounts.remove_file(&p).await.map_err(errno) + } + fn read_via_stream( + &mut self, + fd: Resource, + offset: types::Filesize, + ) -> FsResult> { + let d = self.fd(&fd)?; + let stream: DynInputStream = Box::new(Reader::new(d.file, offset)); + Ok(self.table.push(stream)?) + } + fn write_via_stream( + &mut self, + fd: Resource, + offset: types::Filesize, + ) -> FsResult> { + let d = self.fd(&fd)?; + let stream: DynOutputStream = Box::new(Writer::new(d.file, Some(offset))); + Ok(self.table.push(stream)?) + } + fn append_via_stream( + &mut self, + fd: Resource, + ) -> FsResult> { + let d = self.fd(&fd)?; + let stream: DynOutputStream = Box::new(Writer::new(d.file, None)); + Ok(self.table.push(stream)?) + } + async fn is_same_object( + &mut self, + a: Resource, + b: Resource, + ) -> wasmtime::Result { + Ok(self.fd(&a)?.path == self.fd(&b)?.path) + } + async fn metadata_hash( + &mut self, + fd: Resource, + ) -> FsResult { + Ok(hash(&self.fd(&fd)?.path)) + } + async fn metadata_hash_at( + &mut self, + fd: Resource, + path_flags: types::PathFlags, + path: String, + ) -> FsResult { + let _ = path_flags; + let p = self.path(&fd, &path)?; + self.adapter.mounts.stat(&p).await.map_err(errno)?; + Ok(hash(&p)) + } +} +impl HostDirectoryEntryStream for View<'_> { + async fn read_directory_entry( + &mut self, + stream: Resource, + ) -> FsResult> { + Ok(self + .table + .get_mut(&Resource::::new_borrow(stream.rep()))? + .0 + .next()) + } + fn drop(&mut self, stream: Resource) -> wasmtime::Result<()> { + self.table + .delete(Resource::::new_own(stream.rep()))?; + Ok(()) + } +} diff --git a/vfs/src/generated.rs b/vfs/src/generated.rs new file mode 100644 index 0000000..66c3565 --- /dev/null +++ b/vfs/src/generated.rs @@ -0,0 +1,132 @@ +//! Read-only generated files. Each open captures an independent finite snapshot. +use crate::*; +use std::{collections::BTreeMap, io::ErrorKind as E}; +type Generator = Arc Result> + Send + Sync>; +/// A tree of generated files. Parent folders are implicit. +#[derive(Default)] +pub struct Generated { + files: BTreeMap, +} +impl Generated { + pub fn insert( + &mut self, + path: &str, + generate: impl Fn() -> Result> + Send + Sync + 'static, + ) { + self.files.insert(resolve("/", path), Arc::new(generate)); + } + fn directory(&self, path: &str) -> bool { + path == "/" + || self + .files + .keys() + .any(|p| p.starts_with(&format!("{path}/"))) + } +} +struct Snapshot(Vec); +fn readonly() -> Result { + Err(error(E::ReadOnlyFilesystem)) +} +#[async_trait] +impl Layer for Generated { + async fn stat(&self, path: &str) -> Result { + if let Some(f) = self.files.get(path) { + return Snapshot(f()?).stat().await; + } + if !self.directory(path) { + return Err(error(E::NotFound)); + } + Ok(Stat { + directory: true, + size: 0, + accessed: None, + modified: None, + }) + } + async fn list(&self, path: &str) -> Result> { + if !self.directory(path) { + return Err(error(if self.files.contains_key(path) { + E::NotADirectory + } else { + E::NotFound + })); + } + let prefix = if path == "/" { + "/".into() + } else { + format!("{path}/") + }; + let mut entries = BTreeMap::new(); + for p in self.files.keys() { + if let Some(rest) = p.strip_prefix(&prefix) { + let (name, directory) = rest + .split_once('/') + .map_or((rest, false), |(name, _)| (name, true)); + entries.insert(name.to_string(), directory); + } + } + Ok(entries + .into_iter() + .map(|(name, directory)| Entry { name, directory }) + .collect()) + } + async fn open(&self, path: &str, o: Open) -> Result> { + if o.write || o.create || o.truncate { + return readonly(); + } + if let Some(f) = self.files.get(path) { + if o.directory { + return Err(error(E::NotADirectory)); + } + return Ok(Arc::new(Snapshot(f()?))); + } + if self.directory(path) { + return Memory::default().open("/", o).await; + } + Err(error(E::NotFound)) + } + async fn mkdir(&self, _: &str) -> Result<()> { + readonly() + } + async fn remove_file(&self, _: &str) -> Result<()> { + readonly() + } + async fn remove_dir(&self, _: &str) -> Result<()> { + readonly() + } + async fn rename(&self, _: &str, _: &str) -> Result<()> { + readonly() + } +} +#[async_trait] +impl File for Snapshot { + async fn stat(&self) -> Result { + Ok(Stat { + directory: false, + size: self.0.len() as u64, + accessed: None, + modified: None, + }) + } + async fn read(&self, offset: u64, length: usize) -> Result> { + let start = usize::try_from(offset) + .unwrap_or(usize::MAX) + .min(self.0.len()); + Ok(self.0[start..start.saturating_add(length).min(self.0.len())].to_vec()) + } + async fn write(&self, _: u64, _: Vec) -> Result { + readonly() + } + async fn append(&self, _: Vec) -> Result { + readonly() + } + async fn set_size(&self, _: u64) -> Result<()> { + readonly() + } + async fn sync(&self) -> Result<()> { + Ok(()) + } + async fn set_times(&self, _: Option, _: Option) -> Result<()> { + readonly() + } +} diff --git a/vfs/src/host.rs b/vfs/src/host.rs new file mode 100644 index 0000000..9a4d849 --- /dev/null +++ b/vfs/src/host.rs @@ -0,0 +1,190 @@ +//! Capability-sandboxed host files. Every system call runs on the blocking pool. +use crate::*; +use cap_std::fs::{Dir, OpenOptions}; +use std::{ + io::{Read, Seek, SeekFrom, Write}, + path::Path, + sync::Mutex, +}; + +/// A host directory bound to one process tracker. +#[derive(Clone)] +pub struct Host { + dir: Arc, + process: Process, +} +impl Host { + /// Open the sandbox root. Call this outside an asynchronous worker. + pub fn new(path: impl AsRef, process: Process) -> Result { + Ok(Self { + dir: Arc::new(Dir::open_ambient_dir(path, cap_std::ambient_authority())?), + process, + }) + } + /// Share a sandbox with a different process tracker. + pub fn for_process(&self, process: Process) -> Self { + Self { + dir: self.dir.clone(), + process, + } + } +} +fn rel(path: &str) -> String { + let p = path.trim_start_matches('/'); + if p.is_empty() { ".".into() } else { p.into() } +} +fn stat(m: cap_std::fs::Metadata) -> Stat { + Stat { + directory: m.is_dir(), + size: m.len(), + accessed: m.accessed().ok().map(|t| t.into_std()), + modified: m.modified().ok().map(|t| t.into_std()), + } +} +struct Handle { + file: Arc>, + append: Arc>, + process: Process, +} +#[async_trait] +impl Layer for Host { + async fn stat(&self, path: &str) -> Result { + let (d, p) = (self.dir.clone(), rel(path)); + self.process.blocking(move || d.metadata(p).map(stat)).await + } + async fn list(&self, path: &str) -> Result> { + let (d, p) = (self.dir.clone(), rel(path)); + self.process + .blocking(move || { + let mut entries = d + .read_dir(p)? + .map(|e| { + let e = e?; + Ok(Entry { + name: e.file_name().to_string_lossy().into_owned(), + directory: e.file_type()?.is_dir(), + }) + }) + .collect::>>()?; + entries.sort_by(|a, b| a.name.cmp(&b.name)); + Ok(entries) + }) + .await + } + async fn open(&self, path: &str, o: Open) -> Result> { + let (d, p) = (self.dir.clone(), rel(path)); + let (file, append) = self + .process + .blocking(move || { + let mut options = OpenOptions::new(); + options + .read(true) + .write(o.write) + .create(o.create) + .create_new(o.create && o.exclusive) + .truncate(o.truncate); + let f = d.open_with(&p, &options)?; + let directory = f.metadata()?.is_dir(); + if o.directory && !directory { + return Err(error(io::ErrorKind::NotADirectory)); + } + let append = if o.write && !directory { + d.open_with(&p, OpenOptions::new().append(true))? + } else { + f.try_clone()? + }; + Ok((f, append)) + }) + .await?; + Ok(Arc::new(Handle { + file: Arc::new(Mutex::new(file)), + append: Arc::new(Mutex::new(append)), + process: self.process.clone(), + })) + } + async fn mkdir(&self, path: &str) -> Result<()> { + let (d, p) = (self.dir.clone(), rel(path)); + self.process.blocking(move || d.create_dir(p)).await + } + async fn remove_file(&self, path: &str) -> Result<()> { + let (d, p) = (self.dir.clone(), rel(path)); + self.process.blocking(move || d.remove_file(p)).await + } + async fn remove_dir(&self, path: &str) -> Result<()> { + let (d, p) = (self.dir.clone(), rel(path)); + self.process.blocking(move || d.remove_dir(p)).await + } + async fn rename(&self, from: &str, to: &str) -> Result<()> { + let (d, a, b) = (self.dir.clone(), rel(from), rel(to)); + self.process.blocking(move || d.rename(a, &d, b)).await + } +} +#[async_trait] +impl File for Handle { + async fn stat(&self) -> Result { + let f = self.file.clone(); + self.process + .blocking(move || f.lock().unwrap().metadata().map(stat)) + .await + } + async fn read(&self, offset: u64, length: usize) -> Result> { + let f = self.file.clone(); + self.process + .blocking(move || { + let mut f = f.lock().unwrap(); + f.seek(SeekFrom::Start(offset))?; + let mut bytes = Vec::new(); + (&mut *f).take(length as u64).read_to_end(&mut bytes)?; + Ok(bytes) + }) + .await + } + async fn write(&self, offset: u64, bytes: Vec) -> Result { + let f = self.file.clone(); + self.process + .blocking(move || { + let mut f = f.lock().unwrap(); + f.seek(SeekFrom::Start(offset))?; + f.write(&bytes) + }) + .await + } + async fn append(&self, bytes: Vec) -> Result { + let f = self.append.clone(); + self.process + .blocking(move || f.lock().unwrap().write(&bytes)) + .await + } + async fn set_size(&self, size: u64) -> Result<()> { + let f = self.file.clone(); + self.process + .blocking(move || f.lock().unwrap().set_len(size)) + .await + } + async fn sync(&self) -> Result<()> { + let f = self.file.clone(); + self.process + .blocking(move || f.lock().unwrap().sync_all()) + .await + } + async fn set_times( + &self, + accessed: Option, + modified: Option, + ) -> Result<()> { + let f = self.file.clone(); + self.process + .blocking(move || { + let f = f.lock().unwrap().try_clone()?.into_std(); + let mut times = std::fs::FileTimes::new(); + if let Some(t) = accessed { + times = times.set_accessed(t); + } + if let Some(t) = modified { + times = times.set_modified(t); + } + f.set_times(times) + }) + .await + } +} diff --git a/vfs/src/lib.rs b/vfs/src/lib.rs index f38233f..b9722cf 100644 --- a/vfs/src/lib.rs +++ b/vfs/src/lib.rs @@ -1 +1,92 @@ -//! A composable virtual filesystem for WASI guests in Wasmtime. +//! Composable filesystem layers, open files, mount routing, and process IO guards. + +pub mod adapter; +pub mod generated; +pub mod host; +pub mod memory; +pub mod mounts; +pub mod process; +mod readonly; +mod streams; + +pub use async_trait::async_trait; +pub use generated::Generated; +pub use host::Host; +pub use memory::Memory; +pub use mounts::Mounts; +pub use process::Process; +use std::{io, sync::Arc, time::SystemTime}; + +/// A filesystem result. Error kinds preserve portable filesystem failures. +pub type Result = io::Result; +/// The type and timestamps of a node. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct Stat { + pub directory: bool, + pub size: u64, + pub accessed: Option, + pub modified: Option, +} +/// One child of a folder. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct Entry { + pub name: String, + pub directory: bool, +} +/// Options for an open operation. Exclusive applies only with create. +#[derive(Clone, Copy, Debug, Default)] +pub struct Open { + pub write: bool, + pub create: bool, + pub exclusive: bool, + pub truncate: bool, + pub directory: bool, +} +/// An open file keeps its identity after rename or unlink. +#[async_trait] +pub trait File: Send + Sync { + async fn stat(&self) -> Result; + async fn read(&self, offset: u64, length: usize) -> Result>; + async fn write(&self, offset: u64, bytes: Vec) -> Result; + async fn append(&self, bytes: Vec) -> Result; + async fn set_size(&self, size: u64) -> Result<()>; + async fn sync(&self) -> Result<()>; + async fn set_times( + &self, + accessed: Option, + modified: Option, + ) -> Result<()>; +} +/// One filesystem layer. Paths are normalized, absolute, and relative to its root. +/// Implementations must retain open file identity and return portable IO errors. +#[async_trait] +pub trait Layer: Send + Sync { + async fn stat(&self, path: &str) -> Result; + async fn list(&self, path: &str) -> Result>; + async fn open(&self, path: &str, options: Open) -> Result>; + async fn mkdir(&self, path: &str) -> Result<()>; + async fn remove_file(&self, path: &str) -> Result<()>; + async fn remove_dir(&self, path: &str) -> Result<()>; + async fn rename(&self, from: &str, to: &str) -> Result<()>; +} + +pub(crate) fn error(kind: io::ErrorKind) -> io::Error { + kind.into() +} +/// Resolve a path in the virtual namespace, including traversal across mounts. +pub fn resolve(base: &str, path: &str) -> String { + let mut parts = Vec::new(); + if !path.starts_with('/') { + parts.extend(base.split('/').filter(|s| !s.is_empty())); + } + for part in path.split('/') { + match part { + "" | "." => {} + ".." => { + parts.pop(); + } + p => parts.push(p), + } + } + format!("/{}", parts.join("/")) +} diff --git a/vfs/src/memory.rs b/vfs/src/memory.rs new file mode 100644 index 0000000..1de68e7 --- /dev/null +++ b/vfs/src/memory.rs @@ -0,0 +1,257 @@ +//! Shared in-memory directories and identity-preserving open files. +use crate::*; +use std::{collections::BTreeMap, io::ErrorKind as E, sync::Mutex}; + +#[derive(Clone)] +enum Node { + Dir, + File(Arc>), +} +struct Data { + bytes: Vec, + accessed: Option, + modified: Option, +} +/// A shareable memory filesystem. Clones refer to the same tree. +#[derive(Clone)] +pub struct Memory { + nodes: Arc>>, +} +impl Default for Memory { + fn default() -> Self { + Self { + nodes: Arc::new(Mutex::new(BTreeMap::from([("/".into(), Node::Dir)]))), + } + } +} +fn parent(path: &str) -> &str { + path.rsplit_once('/') + .map(|(p, _)| if p.is_empty() { "/" } else { p }) + .unwrap_or("/") +} +fn directory(nodes: &BTreeMap, path: &str) -> Result<()> { + match nodes.get(path) { + Some(Node::Dir) => Ok(()), + Some(_) => Err(error(E::NotADirectory)), + None => Err(error(E::NotFound)), + } +} +fn metadata(node: &Node) -> Stat { + match node { + Node::Dir => Stat { + directory: true, + size: 0, + accessed: None, + modified: None, + }, + Node::File(data) => { + let d = data.lock().unwrap(); + Stat { + directory: false, + size: d.bytes.len() as u64, + accessed: d.accessed, + modified: d.modified, + } + } + } +} +struct Handle { + node: Node, + write: bool, +} +#[async_trait] +impl Layer for Memory { + async fn stat(&self, path: &str) -> Result { + self.nodes + .lock() + .unwrap() + .get(path) + .map(metadata) + .ok_or_else(|| error(E::NotFound)) + } + async fn list(&self, path: &str) -> Result> { + let nodes = self.nodes.lock().unwrap(); + directory(&nodes, path)?; + Ok(nodes + .iter() + .filter(|(p, _)| p.as_str() != path && parent(p) == path) + .map(|(p, n)| Entry { + name: p.rsplit('/').next().unwrap().into(), + directory: matches!(n, Node::Dir), + }) + .collect()) + } + async fn open(&self, path: &str, o: Open) -> Result> { + let mut nodes = self.nodes.lock().unwrap(); + if o.truncate && !o.write { + return Err(error(E::InvalidInput)); + } + if nodes.contains_key(path) && o.create && o.exclusive { + return Err(error(E::AlreadyExists)); + } + if !nodes.contains_key(path) { + if !o.create || o.directory { + return Err(error(E::NotFound)); + } + directory(&nodes, parent(path))?; + nodes.insert( + path.into(), + Node::File(Arc::new(Mutex::new(Data { + bytes: vec![], + accessed: None, + modified: None, + }))), + ); + } + let node = nodes[path].clone(); + match &node { + Node::Dir if o.write || o.truncate => return Err(error(E::IsADirectory)), + Node::File(_) if o.directory => return Err(error(E::NotADirectory)), + Node::File(d) if o.truncate => d.lock().unwrap().bytes.clear(), + _ => {} + } + Ok(Arc::new(Handle { + node, + write: o.write, + })) + } + async fn mkdir(&self, path: &str) -> Result<()> { + let mut nodes = self.nodes.lock().unwrap(); + if nodes.contains_key(path) { + return Err(error(E::AlreadyExists)); + } + directory(&nodes, parent(path))?; + nodes.insert(path.into(), Node::Dir); + Ok(()) + } + async fn remove_file(&self, path: &str) -> Result<()> { + let mut nodes = self.nodes.lock().unwrap(); + match nodes.get(path) { + None => return Err(error(E::NotFound)), + Some(Node::Dir) => return Err(error(E::IsADirectory)), + _ => {} + } + nodes.remove(path); + Ok(()) + } + async fn remove_dir(&self, path: &str) -> Result<()> { + let mut nodes = self.nodes.lock().unwrap(); + directory(&nodes, path)?; + if path == "/" { + return Err(error(E::ResourceBusy)); + } + if nodes.keys().any(|p| p != path && parent(p) == path) { + return Err(error(E::DirectoryNotEmpty)); + } + nodes.remove(path); + Ok(()) + } + async fn rename(&self, from: &str, to: &str) -> Result<()> { + let mut nodes = self.nodes.lock().unwrap(); + let node = nodes.get(from).cloned().ok_or_else(|| error(E::NotFound))?; + directory(&nodes, parent(to))?; + if from == to { + return Ok(()); + } + if from == "/" || to == "/" { + return Err(error(E::ResourceBusy)); + } + if to.starts_with(&format!("{from}/")) { + return Err(error(E::InvalidInput)); + } + if let Some(target) = nodes.get(to) { + match (&node, target) { + (Node::Dir, Node::File(_)) => return Err(error(E::NotADirectory)), + (Node::File(_), Node::Dir) => return Err(error(E::IsADirectory)), + (Node::Dir, Node::Dir) if nodes.keys().any(|p| p != to && parent(p) == to) => { + return Err(error(E::DirectoryNotEmpty)); + } + _ => {} + } + } + let prefix = format!("{from}/"); + let moved: Vec<_> = nodes + .keys() + .filter(|p| p.as_str() == from || p.starts_with(&prefix)) + .cloned() + .collect(); + for p in moved { + let n = nodes.remove(&p).unwrap(); + nodes.insert(format!("{to}{}", &p[from.len()..]), n); + } + Ok(()) + } +} +impl Handle { + fn data(&self) -> Result<&Mutex> { + match &self.node { + Node::File(d) => Ok(d), + _ => Err(error(E::IsADirectory)), + } + } + fn writable(&self) -> Result<&Mutex> { + if !self.write { + return Err(error(E::PermissionDenied)); + } + self.data() + } +} +#[async_trait] +impl File for Handle { + async fn stat(&self) -> Result { + Ok(metadata(&self.node)) + } + async fn read(&self, offset: u64, length: usize) -> Result> { + let d = self.data()?.lock().unwrap(); + let start = usize::try_from(offset) + .unwrap_or(usize::MAX) + .min(d.bytes.len()); + Ok(d.bytes[start..start.saturating_add(length).min(d.bytes.len())].to_vec()) + } + async fn write(&self, offset: u64, bytes: Vec) -> Result { + let mut d = self.writable()?.lock().unwrap(); + let start = usize::try_from(offset).map_err(|_| error(E::InvalidInput))?; + let end = start + .checked_add(bytes.len()) + .ok_or_else(|| error(E::InvalidInput))?; + if !bytes.is_empty() { + if end > d.bytes.len() { + d.bytes.resize(end, 0); + } + d.bytes[start..end].copy_from_slice(&bytes); + } + Ok(bytes.len()) + } + async fn append(&self, bytes: Vec) -> Result { + self.writable()? + .lock() + .unwrap() + .bytes + .extend_from_slice(&bytes); + Ok(bytes.len()) + } + async fn set_size(&self, size: u64) -> Result<()> { + self.writable()?.lock().unwrap().bytes.resize( + usize::try_from(size).map_err(|_| error(E::InvalidInput))?, + 0, + ); + Ok(()) + } + async fn sync(&self) -> Result<()> { + Ok(()) + } + async fn set_times( + &self, + accessed: Option, + modified: Option, + ) -> Result<()> { + let mut d = self.writable()?.lock().unwrap(); + if accessed.is_some() { + d.accessed = accessed; + } + if modified.is_some() { + d.modified = modified; + } + Ok(()) + } +} diff --git a/vfs/src/mounts.rs b/vfs/src/mounts.rs new file mode 100644 index 0000000..3c474b5 --- /dev/null +++ b/vfs/src/mounts.rs @@ -0,0 +1,97 @@ +//! Resolves virtual paths before routing them to mounted layers. +use crate::*; +use std::{collections::BTreeMap, io::ErrorKind as E}; +struct Mount { + layer: Arc, + readonly: bool, +} +/// A per-process mount namespace. Mount the root before use. +#[derive(Default)] +pub struct Mounts { + mounts: BTreeMap, +} +impl Mounts { + /// Add or replace a mount. Read-only mounts reject every mutation. + pub fn mount(&mut self, path: &str, layer: Arc, readonly: bool) { + self.mounts + .insert(resolve("/", path), Mount { layer, readonly }); + } + fn route(&self, path: &str, write: bool) -> Result<(&Mount, String)> { + let path = resolve("/", path); + let (base, m) = self + .mounts + .iter() + .filter(|(p, _)| p.as_str() == "/" || path == **p || path.starts_with(&format!("{p}/"))) + .max_by_key(|(p, _)| p.len()) + .ok_or_else(|| error(E::NotFound))?; + if write && m.readonly { + return Err(error(E::ReadOnlyFilesystem)); + } + let rest = if base == "/" { + path + } else { + path[base.len()..].into() + }; + Ok((m, if rest.is_empty() { "/".into() } else { rest })) + } +} +#[async_trait] +impl Layer for Mounts { + async fn stat(&self, path: &str) -> Result { + let (m, p) = self.route(path, false)?; + m.layer.stat(&p).await + } + async fn list(&self, path: &str) -> Result> { + let path = resolve("/", path); + let (m, p) = self.route(&path, false)?; + let mut entries = m.layer.list(&p).await?; + let prefix = if path == "/" { + "/".into() + } else { + format!("{path}/") + }; + for mount in self.mounts.keys() { + if let Some(rest) = mount.strip_prefix(&prefix) + && !rest.is_empty() + { + let name = rest.split('/').next().unwrap(); + entries.retain(|e| e.name != name); + entries.push(Entry { + name: name.into(), + directory: true, + }); + } + } + entries.sort_by(|a, b| a.name.cmp(&b.name)); + Ok(entries) + } + async fn open(&self, path: &str, options: Open) -> Result> { + let (m, p) = self.route(path, options.write || options.create || options.truncate)?; + let file = m.layer.open(&p, options).await?; + if m.readonly { + Ok(Arc::new(crate::readonly::Readonly(file))) + } else { + Ok(file) + } + } + async fn mkdir(&self, path: &str) -> Result<()> { + let (m, p) = self.route(path, true)?; + m.layer.mkdir(&p).await + } + async fn remove_file(&self, path: &str) -> Result<()> { + let (m, p) = self.route(path, true)?; + m.layer.remove_file(&p).await + } + async fn remove_dir(&self, path: &str) -> Result<()> { + let (m, p) = self.route(path, true)?; + m.layer.remove_dir(&p).await + } + async fn rename(&self, from: &str, to: &str) -> Result<()> { + let (a, p) = self.route(from, true)?; + let (b, q) = self.route(to, true)?; + if !std::ptr::eq(a, b) { + return Err(error(E::CrossesDevices)); + } + a.layer.rename(&p, &q).await + } +} diff --git a/vfs/src/process.rs b/vfs/src/process.rs new file mode 100644 index 0000000..c71e4d0 --- /dev/null +++ b/vfs/src/process.rs @@ -0,0 +1,51 @@ +//! Tracks blocking operations until their system calls return, independent of cancellation. +use std::sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, +}; +use tokio::sync::Notify; + +#[derive(Default)] +struct State { + count: AtomicUsize, + idle: Notify, +} +/// A per-process operation tracker. Stop the process before awaiting idle. +#[derive(Clone, Default)] +pub struct Process(Arc); +struct Guard(Arc); +impl Drop for Guard { + fn drop(&mut self) { + if self.0.count.fetch_sub(1, Ordering::SeqCst) == 1 { + self.0.idle.notify_waiters(); + } + } +} +impl Process { + /// Run a system call on the blocking pool. Cancellation does not release its guard. + pub async fn blocking( + &self, + operation: impl FnOnce() -> std::io::Result + Send + 'static, + ) -> std::io::Result { + self.0.count.fetch_add(1, Ordering::SeqCst); + let guard = Guard(self.0.clone()); + tokio::task::spawn_blocking(move || { + let _guard = guard; + operation() + }) + .await + .map_err(std::io::Error::other)? + } + /// Wait for all outstanding system calls. The caller must prevent new operations. + pub async fn idle(&self) { + loop { + let notified = self.0.idle.notified(); + tokio::pin!(notified); + notified.as_mut().enable(); + if self.0.count.load(Ordering::SeqCst) == 0 { + return; + } + notified.await; + } + } +} diff --git a/vfs/src/readonly.rs b/vfs/src/readonly.rs new file mode 100644 index 0000000..e4945e3 --- /dev/null +++ b/vfs/src/readonly.rs @@ -0,0 +1,30 @@ +//! Preserves read-only mount errors for mutations through open descriptors. +use crate::*; +pub(crate) struct Readonly(pub Arc); +fn denied() -> Result { + Err(error(io::ErrorKind::ReadOnlyFilesystem)) +} +#[async_trait] +impl File for Readonly { + async fn stat(&self) -> Result { + self.0.stat().await + } + async fn read(&self, offset: u64, length: usize) -> Result> { + self.0.read(offset, length).await + } + async fn write(&self, _: u64, _: Vec) -> Result { + denied() + } + async fn append(&self, _: Vec) -> Result { + denied() + } + async fn set_size(&self, _: u64) -> Result<()> { + denied() + } + async fn sync(&self) -> Result<()> { + self.0.sync().await + } + async fn set_times(&self, _: Option, _: Option) -> Result<()> { + denied() + } +} diff --git a/vfs/src/streams.rs b/vfs/src/streams.rs new file mode 100644 index 0000000..8c2495b --- /dev/null +++ b/vfs/src/streams.rs @@ -0,0 +1,134 @@ +//! Nonblocking WASI streams that await layer operations through readiness. +use crate::*; +use bytes::Bytes; +use std::{future::Future, pin::Pin}; +type Pending = Pin> + Send>>; +use wasmtime_wasi::p2::{InputStream, OutputStream, Pollable, StreamError, StreamResult}; +fn failure(e: io::Error) -> StreamError { + StreamError::LastOperationFailed(e.into()) +} +pub(crate) struct Reader { + file: Arc, + offset: u64, + pending: Option>>, + data: Bytes, + eof: bool, + error: Option, +} +impl Reader { + pub fn new(file: Arc, offset: u64) -> Self { + Self { + file, + offset, + pending: None, + data: Bytes::new(), + eof: false, + error: None, + } + } + fn start(&mut self) { + if self.pending.is_none() && self.data.is_empty() && !self.eof { + let f = self.file.clone(); + let offset = self.offset; + self.pending = Some(Box::pin(async move { f.read(offset, 65536).await })); + } + } +} +#[async_trait] +impl Pollable for Reader { + async fn ready(&mut self) { + self.start(); + if let Some(task) = &mut self.pending { + match task.await { + Ok(data) => { + self.eof = data.is_empty(); + self.offset += data.len() as u64; + self.data = data.into(); + } + Err(e) => self.error = Some(e), + } + self.pending = None; + } + } +} +#[async_trait] +impl InputStream for Reader { + fn read(&mut self, size: usize) -> StreamResult { + if let Some(e) = self.error.take() { + return Err(failure(e)); + } + if self.eof && self.data.is_empty() { + return Err(StreamError::Closed); + } + let n = size.min(self.data.len()); + let data = self.data.split_to(n); + self.start(); + Ok(data) + } +} +pub(crate) struct Writer { + file: Arc, + offset: Option, + pending: Option>, + error: Option, +} +impl Writer { + pub fn new(file: Arc, offset: Option) -> Self { + Self { + file, + offset, + pending: None, + error: None, + } + } +} +#[async_trait] +impl Pollable for Writer { + async fn ready(&mut self) { + if let Some(task) = &mut self.pending { + match task.await { + Ok(n) => { + if let Some(p) = &mut self.offset { + *p += n as u64 + } + } + Err(e) => self.error = Some(e), + } + self.pending = None; + } + } +} +#[async_trait] +impl OutputStream for Writer { + fn write(&mut self, bytes: Bytes) -> StreamResult<()> { + if self.check_write()? == 0 { + return Err(StreamError::trap("write without a permit")); + } + let f = self.file.clone(); + let offset = self.offset; + self.pending = Some(Box::pin(async move { + let mut done = 0; + while done < bytes.len() { + let n = match offset { + Some(p) => f.write(p + done as u64, bytes[done..].to_vec()).await?, + None => f.append(bytes[done..].to_vec()).await?, + }; + if n == 0 { + return Err(error(io::ErrorKind::WriteZero)); + } + done += n; + } + Ok(done) + })); + Ok(()) + } + fn flush(&mut self) -> StreamResult<()> { + Ok(()) + } + fn check_write(&mut self) -> StreamResult { + if let Some(e) = self.error.take() { + return Err(failure(e)); + } + Ok(if self.pending.is_some() { 0 } else { 65536 }) + } +} diff --git a/vfs/tests/guests/files/Cargo.toml b/vfs/tests/guests/files/Cargo.toml new file mode 100644 index 0000000..0ae7c0a --- /dev/null +++ b/vfs/tests/guests/files/Cargo.toml @@ -0,0 +1,8 @@ +[package] +name = "vfs-files-guest" +version = "0.1.0" +edition = "2024" +[workspace] +[dependencies] +tempfile = "3" +rusqlite = { version = "0.38", features = ["bundled"] } diff --git a/vfs/tests/guests/files/src/main.rs b/vfs/tests/guests/files/src/main.rs new file mode 100644 index 0000000..8205b23 --- /dev/null +++ b/vfs/tests/guests/files/src/main.rs @@ -0,0 +1,66 @@ +//! Exercises Rust filesystem clients through the virtual WASI filesystem. +use std::{ + fs::{self, OpenOptions}, + io::{Read, Seek, SeekFrom, Write}, +}; +fn main() { + fs::create_dir("/work").unwrap(); + fs::write("/work/a", b"hello").unwrap(); + let mut f = OpenOptions::new() + .read(true) + .write(true) + .open("/work/a") + .unwrap(); + f.seek(SeekFrom::Start(1)).unwrap(); + f.write_all(b"A").unwrap(); + f.set_len(4).unwrap(); + f.sync_all().unwrap(); + OpenOptions::new() + .append(true) + .open("/work/a") + .unwrap() + .write_all(b"!") + .unwrap(); + assert_eq!(fs::read("/work/a").unwrap(), b"hAll!"); + assert_eq!(fs::metadata("/work/a").unwrap().len(), 5); + fs::rename("/work/a", "/work/b").unwrap(); + assert_eq!(fs::read("/work/../work/b").unwrap(), b"hAll!"); + assert_eq!(fs::read("work/b").unwrap(), b"hAll!"); + assert_eq!(fs::read_dir("/work").unwrap().count(), 1); + assert_eq!( + fs::write("/proc/no", b"no").unwrap_err().kind(), + std::io::ErrorKind::ReadOnlyFilesystem + ); + assert_eq!( + fs::rename("/work/b", "/run/b").unwrap_err().kind(), + std::io::ErrorKind::CrossesDevices + ); + let root = fs::read_dir("/") + .unwrap() + .map(|e| e.unwrap().file_name()) + .collect::>(); + for name in ["proc", "run", "self"] { + assert!(root.iter().any(|n| n == name)); + } + assert_eq!(fs::read("/proc/value").unwrap(), b"snapshot"); + fs::write("/run/test", b"memory").unwrap(); + assert_eq!(fs::read("/run/test").unwrap(), b"memory"); + fs::remove_file("/work/b").unwrap(); + fs::remove_dir("/work").unwrap(); + // WASI std does not implement env::temp_dir, so select the folder explicitly. + let mut temp = tempfile::tempfile_in("/tmp").unwrap(); + temp.write_all(b"temporary").unwrap(); + temp.seek(SeekFrom::Start(0)).unwrap(); + let mut text = String::new(); + temp.read_to_string(&mut text).unwrap(); + assert_eq!(text, "temporary"); + let db = rusqlite::Connection::open("/database.sqlite").unwrap(); + db.execute_batch("CREATE TABLE data (value TEXT); INSERT INTO data VALUES ('row');") + .unwrap(); + drop(db); + let db = rusqlite::Connection::open("/database.sqlite").unwrap(); + let row: String = db + .query_row("SELECT value FROM data", [], |r| r.get(0)) + .unwrap(); + assert_eq!(row, "row"); +} diff --git a/vfs/tests/layers.rs b/vfs/tests/layers.rs new file mode 100644 index 0000000..9312349 --- /dev/null +++ b/vfs/tests/layers.rs @@ -0,0 +1,233 @@ +//! Layer contracts, a memory model, and differential host operation sequences. +use proptest::prelude::*; +use sausage_vfs::*; +use std::{ + io::ErrorKind, + sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, + }, +}; +fn rw() -> Open { + Open { + write: true, + create: true, + ..Open::default() + } +} +async fn contract(layer: &dyn Layer) { + layer.mkdir("/dir").await.unwrap(); + let f = layer.open("/dir/a", rw()).await.unwrap(); + f.write(2, b"abc".to_vec()).await.unwrap(); + f.append(b"d".to_vec()).await.unwrap(); + assert_eq!(f.read(0, 100).await.unwrap(), b"\0\0abcd"); + f.set_size(3).await.unwrap(); + assert_eq!(f.stat().await.unwrap().size, 3); + f.sync().await.unwrap(); + let timestamp = std::time::UNIX_EPOCH + std::time::Duration::from_secs(12345); + f.set_times(Some(timestamp), Some(timestamp)).await.unwrap(); + assert_eq!(f.stat().await.unwrap().modified, Some(timestamp)); + assert_eq!( + layer + .open( + "/dir/a", + Open { + exclusive: true, + ..rw() + } + ) + .await + .err() + .unwrap() + .kind(), + ErrorKind::AlreadyExists + ); + assert_eq!( + layer + .open( + "/dir/a", + Open { + directory: true, + ..Open::default() + } + ) + .await + .err() + .unwrap() + .kind(), + ErrorKind::NotADirectory + ); + layer.rename("/dir/a", "/dir/b").await.unwrap(); + assert_eq!(f.read(2, 1).await.unwrap(), b"a"); + assert_eq!(layer.list("/dir").await.unwrap()[0].name, "b"); + layer.remove_file("/dir/b").await.unwrap(); + assert_eq!(f.read(2, 1).await.unwrap(), b"a"); + layer.remove_dir("/dir").await.unwrap(); +} +#[tokio::test] +async fn memory_contract() { + contract(&Memory::default()).await; +} +#[tokio::test] +async fn host_contract_and_sandbox() { + let tmp = tempfile::tempdir().unwrap(); + let h = Host::new(tmp.path(), Process::default()).unwrap(); + contract(&h).await; + assert!(h.open("/../escape", rw()).await.is_err()); +} +#[tokio::test] +async fn shared_memory_handles_append_atomically() { + let memory = Memory::default(); + let other = memory.clone(); + let first = memory.open("/shared", rw()).await.unwrap(); + let second = other.open("/shared", rw()).await.unwrap(); + let a = tokio::spawn(async move { + for _ in 0..100 { + first.append(b"a".to_vec()).await.unwrap(); + } + }); + let b = tokio::spawn(async move { + for _ in 0..100 { + second.append(b"b".to_vec()).await.unwrap(); + } + }); + a.await.unwrap(); + b.await.unwrap(); + let bytes = memory + .open("/shared", Open::default()) + .await + .unwrap() + .read(0, 1000) + .await + .unwrap(); + assert_eq!(bytes.len(), 200); + assert_eq!(bytes.iter().filter(|b| **b == b'a').count(), 100); +} +#[tokio::test] +async fn generated_snapshots() { + let counter = Arc::new(AtomicUsize::new(0)); + let c = counter.clone(); + let mut g = Generated::default(); + g.insert("/nested/data", move || { + Ok(c.fetch_add(1, Ordering::SeqCst).to_string().into_bytes()) + }); + let a = g.open("/nested/data", Open::default()).await.unwrap(); + let b = g.open("/nested/data", Open::default()).await.unwrap(); + assert_eq!(a.read(0, 99).await.unwrap(), b"0"); + assert_eq!(b.read(0, 99).await.unwrap(), b"1"); + assert!(a.read(1, 99).await.unwrap().is_empty()); + assert_eq!(g.list("/").await.unwrap()[0].name, "nested"); + assert_eq!( + g.remove_file("/nested/data").await.unwrap_err().kind(), + ErrorKind::ReadOnlyFilesystem + ); +} +#[tokio::test] +async fn mounts_resolve_and_protect() { + let mut mounts = Mounts::default(); + mounts.mount("/", Arc::new(Memory::default()), false); + mounts.mount("/proc", Arc::new(Memory::default()), true); + mounts.mount("/run", Arc::new(Memory::default()), false); + assert_eq!( + mounts + .list("/") + .await + .unwrap() + .iter() + .map(|e| e.name.as_str()) + .collect::>(), + ["proc", "run"] + ); + mounts.open("/proc/../a", rw()).await.unwrap(); + assert_eq!( + mounts.open("/proc/a", rw()).await.err().unwrap().kind(), + ErrorKind::ReadOnlyFilesystem + ); + assert_eq!( + mounts.rename("/a", "/run/a").await.unwrap_err().kind(), + ErrorKind::CrossesDevices + ); +} +#[tokio::test] +async fn aborted_operations_keep_guard() { + let p = Process::default(); + let q = p.clone(); + let (started_tx, started_rx) = tokio::sync::oneshot::channel(); + let (release_tx, release_rx) = std::sync::mpsc::channel(); + let operation = tokio::spawn(async move { + q.blocking(move || { + started_tx.send(()).unwrap(); + release_rx.recv().unwrap(); + Ok(()) + }) + .await + }); + started_rx.await.unwrap(); + operation.abort(); + let mut idle = Box::pin(p.idle()); + assert!(matches!( + std::future::poll_fn(|cx| std::task::Poll::Ready(idle.as_mut().poll(cx))).await, + std::task::Poll::Pending + )); + release_tx.send(()).unwrap(); + idle.await; +} +proptest! { + #[test] + fn memory_model(ops in prop::collection::vec((0u8..3,0usize..64,prop::collection::vec(any::(),0..20)),0..80)) { + tokio::runtime::Runtime::new().unwrap().block_on(async { + let m = Memory::default(); let f = m.open("/f",rw()).await.unwrap(); let mut model = Vec::new(); + for (op,offset,bytes) in ops { + match op { + 0 => { f.write(offset as u64,bytes.clone()).await.unwrap(); if !bytes.is_empty() { model.resize(model.len().max(offset+bytes.len()),0); model[offset..offset+bytes.len()].copy_from_slice(&bytes); } }, + 1 => { f.append(bytes.clone()).await.unwrap(); model.extend(bytes); }, + _ => { f.set_size(offset as u64).await.unwrap(); model.resize(offset,0); } + } + prop_assert_eq!(f.read(0,1000).await.unwrap(),model.clone()); + } + Ok(()) + })?; + } + #[test] + fn host_differential(ops in prop::collection::vec((0u8..8,0u8..5,prop::collection::vec(any::(),0..30)),0..50)) { + tokio::runtime::Runtime::new().unwrap().block_on(async { + let a = tempfile::tempdir().unwrap(); let b = tempfile::tempdir().unwrap(); let host = Host::new(a.path(),Process::default()).unwrap(); + for (op,index,data) in ops { + let name = format!("f{index}"); let path = format!("/{name}"); let real = b.path().join(&name); + let (actual,expected) = match op { + 0 => { let actual = match host.open(&path,Open { truncate:true,..rw() }).await { Ok(f) => f.write(0,data.clone()).await.map(|_|()), Err(e) => Err(e) }; (actual,std::fs::write(real,&data)) }, + 1 => (host.remove_file(&path).await,std::fs::remove_file(real)), + 2 => (host.mkdir(&path).await,std::fs::create_dir(real)), + 3 => (host.remove_dir(&path).await,std::fs::remove_dir(real)), + 4 => (host.rename(&path,"/renamed").await,std::fs::rename(real,b.path().join("renamed"))), + _ => { + use std::io::{Seek, SeekFrom, Write}; + let actual = match host.open(&path,Open { write:true,..Open::default() }).await { + Ok(f) => match op { 5 => f.append(data.clone()).await.map(|_|()), 6 => f.set_size(data.len() as u64).await, _ => f.write(index as u64,data.clone()).await.map(|_|()) }, + Err(e) => Err(e), + }; + let expected = (|| { + let mut f = std::fs::OpenOptions::new().read(true).write(true).append(op == 5).open(real)?; + match op { 5 => f.write_all(&data), 6 => f.set_len(data.len() as u64), _ => { f.seek(SeekFrom::Start(index as u64))?; f.write_all(&data) } } + })(); + (actual,expected) + } + }; + prop_assert_eq!(actual.map_err(|e|e.kind()),expected.map_err(|e|e.kind())); + let mut expected = std::fs::read_dir(b.path()).unwrap().map(|e| { let e=e.unwrap(); Entry { name:e.file_name().to_str().unwrap().into(),directory:e.file_type().unwrap().is_dir() } }).collect::>(); expected.sort_by(|a,b|a.name.cmp(&b.name)); + for entry in &expected { + let p = format!("/{}", entry.name); + let metadata = host.stat(&p).await.unwrap(); + prop_assert_eq!(metadata.directory, entry.directory); + if !entry.directory { + let bytes = std::fs::read(b.path().join(&entry.name)).unwrap(); + prop_assert_eq!(metadata.size, bytes.len() as u64); + prop_assert_eq!(host.open(&p, Open::default()).await.unwrap().read(0, 1024).await.unwrap(), bytes); + } + } + prop_assert_eq!(host.list("/").await.unwrap(),expected); + } + Ok(()) + })?; + } +} diff --git a/vfs/tests/streams.rs b/vfs/tests/streams.rs new file mode 100644 index 0000000..0221088 --- /dev/null +++ b/vfs/tests/streams.rs @@ -0,0 +1,67 @@ +//! Exercises adapter streams with the shared WASI resource table. +use sausage_vfs::{ + adapter::{Adapter, View}, + *, +}; +use std::sync::Arc; +use wasmtime::component::{Resource, ResourceTable}; +use wasmtime_wasi::p2::{ + StreamError, + bindings::filesystem::{ + preopens::Host, + types::{DescriptorFlags, HostDescriptor, OpenFlags, PathFlags}, + }, +}; +#[tokio::test] +async fn adapter_streams() { + let mut mounts = Mounts::default(); + mounts.mount("/", Arc::new(Memory::default()), false); + let mut adapter = Adapter::new(mounts); + let mut table = ResourceTable::new(); + let mut view = View { + adapter: &mut adapter, + table: &mut table, + }; + let root = view.get_directories().unwrap().remove(0).0; + let fd = view + .open_at( + root, + PathFlags::empty(), + "file".into(), + OpenFlags::CREATE, + DescriptorFlags::READ | DescriptorFlags::WRITE, + ) + .await + .unwrap(); + let output = view + .write_via_stream(Resource::new_borrow(fd.rep()), 2) + .unwrap(); + { + let stream = view.table.get_mut(&output).unwrap(); + assert!(stream.check_write().unwrap() > 0); + stream.write(b"abc".as_slice().to_vec().into()).unwrap(); + stream.flush().unwrap(); + stream.ready().await; + assert!(stream.check_write().unwrap() > 0); + } + let append = view + .append_via_stream(Resource::new_borrow(fd.rep())) + .unwrap(); + { + let stream = view.table.get_mut(&append).unwrap(); + stream.write(b"!".as_slice().to_vec().into()).unwrap(); + stream.ready().await; + } + let input = view + .read_via_stream(Resource::new_borrow(fd.rep()), 0) + .unwrap(); + let stream = view.table.get_mut(&input).unwrap(); + assert_eq!( + stream.blocking_read(100).await.unwrap().as_ref(), + b"\0\0abc!" + ); + assert!(matches!( + stream.blocking_read(100).await, + Err(StreamError::Closed) + )); +} diff --git a/vfs/tests/wasi.rs b/vfs/tests/wasi.rs new file mode 100644 index 0000000..0527e3b --- /dev/null +++ b/vfs/tests/wasi.rs @@ -0,0 +1,94 @@ +//! Compiles a real WASIp2 guest and runs std, SQLite, and tempfile through the adapter. +use sausage_vfs::{ + adapter::{self, Adapter, View}, + *, +}; +use std::sync::Arc; +use wasmtime::{ + Engine, Store, + component::{Component, Linker, ResourceTable}, +}; +use wasmtime_wasi::{WasiCtx, WasiCtxBuilder, WasiCtxView, WasiView}; +struct State { + wasi: WasiCtx, + table: ResourceTable, + fs: Adapter, +} +impl WasiView for State { + fn ctx(&mut self) -> WasiCtxView<'_> { + WasiCtxView { + ctx: &mut self.wasi, + table: &mut self.table, + } + } +} +#[tokio::test(flavor = "multi_thread")] +async fn rust_guest() { + let manifest = + std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("tests/guests/files/Cargo.toml"); + let target = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("../target/vfs/guests"); + let status = std::process::Command::new("cargo") + .args([ + "build", + "--release", + "--locked", + "--offline", + "--target", + "wasm32-wasip2", + "--manifest-path", + ]) + .arg(manifest) + .env("CARGO_TARGET_DIR", &target) + .status() + .unwrap(); + assert!(status.success()); + let world = tempfile::tempdir().unwrap(); + std::fs::create_dir(world.path().join("tmp")).unwrap(); + let mut mounts = Mounts::default(); + mounts.mount( + "/", + Arc::new(Host::new(world.path(), Process::default()).unwrap()), + false, + ); + mounts.mount("/run", Arc::new(Memory::default()), false); + mounts.mount("/self", Arc::new(Memory::default()), false); + let mut generated = Generated::default(); + generated.insert("/value", || Ok(b"snapshot".to_vec())); + mounts.mount("/proc", Arc::new(generated), true); + let engine = Engine::default(); + let mut linker = Linker::new(&engine); + wasmtime_wasi::p2::add_to_linker_async(&mut linker).unwrap(); + adapter::add_to_linker(&mut linker, |s: &mut State| View { + adapter: &mut s.fs, + table: &mut s.table, + }) + .unwrap(); + let component = Component::from_file( + &engine, + target.join("wasm32-wasip2/release/vfs-files-guest.wasm"), + ) + .unwrap(); + let mut store = Store::new( + &engine, + State { + wasi: WasiCtxBuilder::new() + .inherit_stdio() + .env("TMPDIR", "/tmp") + .build(), + table: ResourceTable::new(), + fs: Adapter::new(mounts), + }, + ); + let command = + wasmtime_wasi::p2::bindings::Command::instantiate_async(&mut store, &component, &linker) + .await + .unwrap(); + assert!( + command + .wasi_cli_run() + .call_run(&mut store) + .await + .unwrap() + .is_ok() + ); +}