diff --git a/Cargo.lock b/Cargo.lock index b9a0d18..5df8602 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2,18 +2,42 @@ # It is not intended for manual editing. version = 4 +[[package]] +name = "ascii" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d92bec98840b8f03a5ff5413de5293bfcd8bf96467cf5452609f939ec6f5de16" + [[package]] name = "base64" version = "0.22.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" +[[package]] +name = "bitflags" +version = "2.13.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3ded4057c258ba199e2d26386d3af3780957ecaee6c4ef4041c6b4b8b97c0b06" + [[package]] name = "bytes" version = "1.12.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" +[[package]] +name = "cfg-if" +version = "1.0.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4e7648175b45a9a48536d676f68d918270699102aa8dab5496df06904c914600" + +[[package]] +name = "chunked_transfer" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6e4de3bc4ea267985becf712dc6d9eed8b04c953b3fcfb339ebc87acd9804901" + [[package]] name = "displaydoc" version = "0.2.7" @@ -31,6 +55,22 @@ version = "1.0.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f" +[[package]] +name = "errno" +version = "0.3.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" +dependencies = [ + "libc", + "windows-sys", +] + +[[package]] +name = "fastrand" +version = "2.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "da7c62ceae207dd37ea5b845da6a0696c799f85e97da1ab5b7910be3c1c80223" + [[package]] name = "form_urlencoded" version = "1.2.2" @@ -40,6 +80,17 @@ dependencies = [ "percent-encoding", ] +[[package]] +name = "getrandom" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "300e883d756b2e4ec94e02791f39b04b522276138852cfc41d9fb7e904106099" +dependencies = [ + "cfg-if", + "libc", + "r-efi", +] + [[package]] name = "glob" version = "0.3.4" @@ -62,6 +113,12 @@ dependencies = [ "itoa", ] +[[package]] +name = "httpdate" +version = "1.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" + [[package]] name = "icu_collections" version = "2.3.0" @@ -182,18 +239,42 @@ version = "1.0.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" +[[package]] +name = "libc" +version = "0.2.189" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2" + +[[package]] +name = "linux-raw-sys" +version = "0.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32a66949e030da00e8c7d4434b251670a91556f4144941d37452769c25d58a53" + [[package]] name = "litemap" version = "0.8.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "47d9d19d1d6efa0109d2f65ff4c85cddd50bd572e5a00127ab10987290bcefae" +[[package]] +name = "log" +version = "0.4.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f9f8bd3e56ce4dfc153cf470fffbfa98c7620958b312ca5c3a4b8d5181fd13c6" + [[package]] name = "memchr" version = "2.8.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" +[[package]] +name = "once_cell" +version = "1.21.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" + [[package]] name = "percent-encoding" version = "2.3.2" @@ -227,6 +308,25 @@ dependencies = [ "proc-macro2", ] +[[package]] +name = "r-efi" +version = "6.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" + +[[package]] +name = "rustix" +version = "1.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "891efababe418670775f199f0d233d84843c227a0949a883ce15b37c78d6629d" +dependencies = [ + "bitflags", + "errno", + "libc", + "linux-raw-sys", + "windows-sys", +] + [[package]] name = "serde" version = "1.0.229" @@ -329,6 +429,19 @@ version = "1.0.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "876fef147edbcbddc8ac5cbbba92c7b86519e314e86638596c09673b2ed01e7f" +[[package]] +name = "tempfile" +version = "3.27.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" +dependencies = [ + "fastrand", + "getrandom", + "once_cell", + "rustix", + "windows-sys", +] + [[package]] name = "termcolor" version = "1.4.1" @@ -338,6 +451,18 @@ dependencies = [ "winapi-util", ] +[[package]] +name = "tiny_http" +version = "0.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "389915df6413a2e74fb181895f933386023c71110878cd0825588928e64cdc82" +dependencies = [ + "ascii", + "chunked_transfer", + "httpdate", + "log", +] + [[package]] name = "tinystr" version = "0.8.4" @@ -456,6 +581,20 @@ version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "23b97319f7b8343df12cc98938e5c3eb436064524c8d2b4e30a1d3a36eecdf81" +[[package]] +name = "wip-fs-server" +version = "0.1.0" +dependencies = [ + "http", + "rustix", + "tempfile", + "tiny_http", + "wip-host", + "wip-https", + "wip-interface", + "wip-protocol", +] + [[package]] name = "wip-host" version = "0.1.0" diff --git a/README.md b/README.md index 8b83f4b..aec875b 100644 --- a/README.md +++ b/README.md @@ -5,12 +5,18 @@ Rust building blocks for the Web Interface Protocol (WIP). The repository is a resolver-3 virtual Cargo workspace whose packages all live under `crates/*`. It provides a transport-independent protocol model, the standard framework-independent WIP over HTTPS v1 binding, a Rust -interface-authoring facade, a request-scoped host runtime, and a private -public-API conformance consumer. The repository root is not itself a package. +interface-authoring facade, a request-scoped host runtime, a bounded Linux +filesystem projection server, and a private public-API conformance consumer. The +repository root is not itself a package. ## Crate graph and responsibilities ```text +wip-fs-server (request-time filesystem projection and loopback listener) +├── wip-https standard routes/codecs/envelopes/batching +├── wip-host request-scoped publication and dispatch +└── wip-interface macro-generated filesystem capabilities + wip-public-api-example (workspace-only executable, integration tests, doctest) ├── wip-protocol logical declarations, values, calls, and validation ├── wip-https standard HTTPS v1 routes, JSON codecs, envelopes, and batching @@ -29,7 +35,8 @@ wip-public-api-example (workspace-only executable, integration tests, doctest) | `wip-https` | Standard HTTPS v1 endpoint/route derivation, UTF-8 JSON codecs, direct success bodies, error/status classification, and retrieval batches. | HTTP I/O, TLS termination, authentication schemes, server frameworks, filesystem projection, or client presentation/cache policy. | | `wip-interface` | The public `WipType`/`interface` authoring API, Rust-to-protocol codecs, generated descriptors, and typed implementation dispatch. | Object publication, backend lifetime, or transport policy. | | `wip-interface-macros` | Procedural-macro expansion used by `wip-interface`. It is not an application dependency. | A supported direct user API. | -| `wip-host` | Opaque interface registration, explicit object membership, validator checks, and interface-qualified dispatch against one host-owned instance. | Backend atomicity, global object lookup, or network serving. | +| `wip-host` | Opaque interface registration, explicit object membership, validator checks, interface-qualified dispatch against one host-owned instance, and descriptor-guided argument decoding inside that same resolution boundary. | Backend atomicity, global object lookup, or network serving. | +| `wip-fs-server` | Linux directory-handle-confined request-time projection, bounded BFS trees and file operations, standard `wip-https` dispatch/batch handling, and a loopback HTTP binary for TLS-proxy or explicit development use. | A new wire profile, implicit symlink following, non-loopback cleartext, authentication policy, or filesystem-wide preregistration. | | `wip-public-api-example` | An independently compiled consumer of the protocol, authoring, and host public crates. Its DTOs and traits are application-owned; its implementation and backend handle remain host-private. | Reusable library API or a production server. | The boundary is intentional: protocol types describe what can cross a @@ -238,6 +245,17 @@ rolled back. Without that proof, post-invocation failures remain comparison and backend commit atomic and return a validator describing the state that becomes committed. +## Filesystem projection server + +[`crates/wip-fs-server`](crates/wip-fs-server) contains the practical server +binary and its full security model, configuration, curl walkthrough, and +resource/failure semantics. It resolves every object from current filesystem +state, uses pinned directory-relative handles without following symlinks, omits +non-UTF-8 backend names, performs bounded breadth-first tree discovery, and +publishes validator-protected replacement only in explicit read-write mode. +The built-in cleartext listener is loopback-only: production HTTPS is explicitly +delegated to a reverse proxy, while local plain HTTP requires an opt-in flag. + ## Validation Enter the dev shell and run the acceptance commands exactly: diff --git a/crates/wip-fs-server/Cargo.toml b/crates/wip-fs-server/Cargo.toml new file mode 100644 index 0000000..e0707f5 --- /dev/null +++ b/crates/wip-fs-server/Cargo.toml @@ -0,0 +1,20 @@ +[package] +name = "wip-fs-server" +version = "0.1.0" +edition = "2024" +description = "Request-time filesystem projection server for WIP over HTTPS" +license = "MIT OR Apache-2.0" +readme = "README.md" +publish = false + +[dependencies] +http = "1" +rustix = { version = "1.1", features = ["fs"] } +tiny_http = "0.12" +wip-host = { path = "../wip-host" } +wip-https = { path = "../wip-https" } +wip-interface = { path = "../wip-interface" } +wip-protocol = { path = "../wip-protocol" } + +[dev-dependencies] +tempfile = "3" diff --git a/crates/wip-fs-server/README.md b/crates/wip-fs-server/README.md new file mode 100644 index 0000000..339688e --- /dev/null +++ b/crates/wip-fs-server/README.md @@ -0,0 +1,136 @@ +# wip-fs-server + +`wip-fs-server` projects one configured Linux filesystem directory onto WIP +Worldspace `/` at request time. It does not register the filesystem contents or +keep one object instance per entry. It uses `wip-host` for publication and typed +operation dispatch and uses the standard `wip-https` routes, JSON codecs, +envelopes, batching, and HTTP status mapping. + +## Security model + +- The effective root is opened once as a directory capability. Every request is + resolved segment-by-segment with directory-relative, `O_NOFOLLOW` handles. +- Symlinks are projected as links and are never followed. Their stored target is + available only through the `SymlinkInspect.read_link` application operation. +- Regular files, directories, symlinks, and unsupported entry types have distinct + public descriptions and capability sets. +- WIP names are Unicode. Linux names that are not valid UTF-8 are deliberately + not published; valid UTF-8 names map byte-for-byte and therefore cannot collide. +- No object `ref` is issued. This avoids claiming representation identity for + hard links or path-specific names. Regular files and symlinks receive an opaque + validator containing filesystem identity plus protected metadata; replacing a + path therefore does not preserve its validator. +- Protocol metadata and errors never contain the backend absolute path. Startup + prints only the explicitly configured effective root and security-sensitive + modes. +- The built-in listener is cleartext and loopback-only. Production deployments + must pass `--behind-tls-proxy` and terminate HTTPS at a trusted reverse proxy. + Plain development HTTP requires the explicit `--development-http` flag. + +The implementation currently targets Linux because it uses `O_PATH`, +directory-relative descriptor traversal, and `/proc/self/fd` to reopen the same +pinned backend object without resolving the untrusted Worldspace path twice. + +## Run + +Read-only behind a TLS-terminating reverse proxy: + +```console +cargo run -p wip-fs-server -- \ + --root /srv/public-files \ + --listen 127.0.0.1:8443 \ + --base-path /wip \ + --behind-tls-proxy +``` + +Local development with the write capability deliberately enabled: + +```console +cargo run -p wip-fs-server -- \ + --root ./example-files \ + --listen 127.0.0.1:8080 \ + --base-path /wip \ + --development-http \ + --read-write +``` + +The server reports the effective root, `ReadOnly`/`ReadWrite` mode, transport +mode, endpoint, and actual listen address. It does not enumerate the root. + +## Standard interactions + +The following examples use development HTTP for readability. The JSON is the +standard `wip-https` profile; validators and operation `Bytes` are base64 body +fields, never `ETag` or `If-Match` headers. + +Fetch root: + +```console +curl -sS -X POST http://127.0.0.1:8080/wip/v1/fetch \ + -H 'Content-Type: application/json' -H 'Accept: application/json' \ + --data '{"path":"/"}' +``` + +Fetch a bounded tree (the root counts toward `limit`): + +```console +curl -sS -X POST http://127.0.0.1:8080/wip/v1/fetch_tree \ + -H 'Content-Type: application/json' -H 'Accept: application/json' \ + --data '{"path":"/","depth":2,"limit":100}' +``` + +A regular-file fetch returns the opaque read interface reference and an object +validator. Retrieve that descriptor with: + +```console +curl -sS -X POST http://127.0.0.1:8080/wip/v1/fetch_interface \ + -H 'Content-Type: application/json' -H 'Accept: application/json' \ + --data '{"interface":""}' +``` + +Then call a bounded range read: + +```console +curl -sS -X POST http://127.0.0.1:8080/wip/v1/call_operation \ + -H 'Content-Type: application/json' -H 'Accept: application/json' \ + --data '{ + "target":{"path":"/notes.txt","validator":""}, + "interface":{"reference":""}, + "operation":"read", + "arguments":{"offset":0,"length":4096} + }' +``` + +In `--read-write` mode a second regular-file interface declares `replace`. +Writes require the current object validator and return the post-write validator: + +```console +curl -sS -X POST http://127.0.0.1:8080/wip/v1/call_operation \ + -H 'Content-Type: application/json' -H 'Accept: application/json' \ + --data '{ + "target":{"path":"/notes.txt","validator":""}, + "interface":{"reference":""}, + "operation":"replace", + "arguments":{"contents":"dXBkYXRlZA=="} + }' +``` + +`fetch`, `fetch_tree`, and `fetch_interface` can also be combined through the +standard `POST /wip/v1/batch` route. `call_operation` is never batchable. + +## Resource and failure behavior + +`ResourceLimits` bounds request/response bytes, JSON nesting, batch items, file +read/write bytes, tree depth/node count, pinned handles, concurrent requests, and +a cooperative deadline. Tree traversal is breadth-first and opens only requested +entries. Once a limit or depth boundary is reached it reads only far enough to +prove whether an additional visible child exists, so `truncated` is true only +when something was actually omitted. I/O failure and cancellation remain errors, +not clean exhaustion. + +Whole-file replacement compares the request validator against the identity and +state of the same pinned file handle used for the write. Failures before mutation +carry a `NotCommitted` guarantee and map to `Internal` or +`ResourceLimitExceeded`; a failure after truncation/write begins maps to +`OperationOutcomeUnknown`. The server and Clients do not automatically retry +side-effecting operations. diff --git a/crates/wip-fs-server/src/interfaces.rs b/crates/wip-fs-server/src/interfaces.rs new file mode 100644 index 0000000..dd69cb4 --- /dev/null +++ b/crates/wip-fs-server/src/interfaces.rs @@ -0,0 +1,43 @@ +//! Filesystem capability interfaces published by the projection. + +use wip_interface as wip; + +/// Bounded reads from a regular file. +#[wip::interface] +pub trait FileRead { + /// Reads bytes from the current regular file. + /// + /// # Returns + /// At most the configured read limit, beginning at `offset`. + fn read( + &self, + /// Byte offset; omitted means zero. + offset: Option, + /// Requested byte count; omitted means the rest of the file. + length: Option, + ) -> wip::OperationResult; +} + +/// Inspection operations for a symbolic link itself. +#[wip::interface] +pub trait SymlinkInspect { + /// Reads the link payload without following it. + /// + /// # Returns + /// The platform path bytes stored in the symbolic link. + fn read_link(&self) -> wip::OperationResult; +} + +/// Explicitly enabled mutation of a regular file. +#[wip::interface] +pub trait FileWrite { + /// Replaces the current regular file contents through its pinned handle. + /// + /// # Returns + /// Unit after the new bytes have been written and synchronized. + fn replace( + &self, + /// Complete replacement contents. + contents: wip::Bytes, + ) -> wip::OperationResult<()>; +} diff --git a/crates/wip-fs-server/src/lib.rs b/crates/wip-fs-server/src/lib.rs new file mode 100644 index 0000000..e4411f1 --- /dev/null +++ b/crates/wip-fs-server/src/lib.rs @@ -0,0 +1,197 @@ +//! Request-time filesystem projection served through the standard WIP HTTPS binding. +//! +//! This crate deliberately keeps the boundaries visible: +//! +//! - [`wip_protocol`] owns logical requests and responses; +//! - [`wip_host`] owns object publication, validators, and typed dispatch; +//! - [`wip_https`] owns routes, JSON, envelopes, batching, and status mapping; +//! - this crate owns the filesystem capability boundary and the real HTTP listener. +//! +//! TLS is terminated by an explicitly configured reverse proxy. The built-in +//! listener accepts cleartext only on loopback, either behind that proxy or when +//! [`TransportSecurity::DevelopmentHttp`] is deliberately selected. + +#![deny(missing_docs)] + +mod interfaces; +mod projection; +mod server; + +use std::net::{IpAddr, Ipv4Addr, SocketAddr}; +use std::path::{Path, PathBuf}; +use std::time::Duration; + +pub use interfaces::{FileRead, FileWrite, SymlinkInspect}; +pub use server::{BoundServer, ServerBuildError, WipFsServer}; + +/// Filesystem capabilities published by regular files. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum AccessMode { + /// Publish bounded reads only. + ReadOnly, + /// Also publish validator-protected whole-file replacement. + ReadWrite, +} + +/// How HTTPS is provided for the endpoint. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum TransportSecurity { + /// A trusted reverse proxy terminates TLS and forwards to the loopback listener. + ReverseProxyTls, + /// Cleartext HTTP is deliberately enabled for local development only. + DevelopmentHttp, +} + +/// Bounded resource policy applied by the HTTP adapter and filesystem provider. +#[derive(Debug, Clone, Copy)] +pub struct ResourceLimits { + /// Standard JSON, body-size, nesting, and batch limits. + pub wire: wip_https::Limits, + /// Maximum bytes returned by one file read. + pub max_file_read_bytes: usize, + /// Maximum bytes accepted by one file replacement. + pub max_file_write_bytes: usize, + /// Maximum accepted `fetch_tree` depth. + pub max_tree_depth: u32, + /// Maximum accepted `fetch_tree` node count, including its root. + pub max_tree_nodes: u32, + /// Maximum simultaneously pinned projected filesystem handles. + pub max_open_handles: usize, + /// Maximum requests executing concurrently. + pub max_concurrent_requests: usize, + /// Deadline applied to request dispatch and cooperative filesystem work. + pub request_timeout: Duration, +} + +impl Default for ResourceLimits { + fn default() -> Self { + Self { + wire: wip_https::Limits::new(256 * 1024, 2 * 1024 * 1024, 64, 64) + .expect("default wire limits are nonzero"), + max_file_read_bytes: 1024 * 1024, + max_file_write_bytes: 1024 * 1024, + max_tree_depth: 32, + max_tree_nodes: 1024, + max_open_handles: 2048, + max_concurrent_requests: 32, + request_timeout: Duration::from_secs(30), + } + } +} + +/// Complete listener and projection configuration. +#[derive(Debug, Clone)] +pub struct ServerConfig { + /// Filesystem directory projected as Worldspace `/`. + pub root: PathBuf, + /// Loopback socket used by the built-in cleartext listener. + pub listen: SocketAddr, + /// URL path prefix below which standard `v1/*` routes are mounted. + pub endpoint_base_path: String, + /// Read-only or explicit read-write publication mode. + pub access: AccessMode, + /// TLS termination or explicit development HTTP policy. + pub transport_security: TransportSecurity, + /// Request, operation, traversal, and concurrency bounds. + pub limits: ResourceLimits, +} + +impl ServerConfig { + /// Builds a secure-by-default configuration for `root`. + /// + /// The listener defaults to an ephemeral loopback port, read-only access, + /// and TLS termination delegated to a reverse proxy. + #[must_use] + pub fn for_root(root: impl Into) -> Self { + Self { + root: root.into(), + listen: SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0), + endpoint_base_path: "/wip".into(), + access: AccessMode::ReadOnly, + transport_security: TransportSecurity::ReverseProxyTls, + limits: ResourceLimits::default(), + } + } + + pub(crate) fn validate(&self) -> Result<(), ConfigError> { + if !self.listen.ip().is_loopback() { + return Err(ConfigError::NonLoopbackCleartext); + } + if !valid_base_path(&self.endpoint_base_path) { + return Err(ConfigError::InvalidBasePath); + } + if self.limits.max_file_read_bytes == 0 + || self.limits.max_file_write_bytes == 0 + || self.limits.max_tree_nodes == 0 + || self.limits.max_open_handles == 0 + || self.limits.max_concurrent_requests == 0 + || self.limits.request_timeout.is_zero() + { + return Err(ConfigError::ZeroLimit); + } + if self.limits.max_open_handles < self.limits.max_tree_nodes as usize { + return Err(ConfigError::InsufficientHandles); + } + if self.limits.wire.max_response_bytes() < 512 { + return Err(ConfigError::ResponseLimitTooSmall); + } + Ok(()) + } + + pub(crate) fn effective_root(&self) -> Result { + std::fs::canonicalize(&self.root) + } + + pub(crate) fn endpoint(&self) -> Result { + let endpoint = format!("https://localhost{}", self.endpoint_base_path); + wip_https::Endpoint::parse(&endpoint).map_err(|_| ConfigError::InvalidBasePath) + } +} + +/// Invalid server configuration. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ConfigError { + /// The cleartext built-in listener was exposed beyond loopback. + NonLoopbackCleartext, + /// The endpoint prefix is not an absolute canonical URL path. + InvalidBasePath, + /// A resource limit is zero. + ZeroLimit, + /// Tree nodes could exhaust the configured handle ceiling by construction. + InsufficientHandles, + /// The response limit cannot reliably carry a bounded protocol error envelope. + ResponseLimitTooSmall, +} + +impl std::fmt::Display for ConfigError { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.write_str(match self { + Self::NonLoopbackCleartext => "the built-in cleartext listener must bind to loopback", + Self::InvalidBasePath => { + "endpoint base path must be absolute, canonical, and query-free" + } + Self::ZeroLimit => "all resource limits and the request timeout must be nonzero", + Self::InsufficientHandles => "max_open_handles must cover max_tree_nodes", + Self::ResponseLimitTooSmall => "max_response_bytes must be at least 512", + }) + } +} + +impl std::error::Error for ConfigError {} + +fn valid_base_path(path: &str) -> bool { + path.starts_with('/') + && !path.contains('?') + && !path.contains('#') + && !path.contains("//") + && !path + .split('/') + .any(|segment| segment == "." || segment == "..") + && (path == "/" || !path.ends_with('/')) +} + +/// Returns whether `root` is an existing directory suitable for projection. +#[must_use] +pub fn is_projectable_root(root: &Path) -> bool { + root.is_dir() +} diff --git a/crates/wip-fs-server/src/main.rs b/crates/wip-fs-server/src/main.rs new file mode 100644 index 0000000..0061cff --- /dev/null +++ b/crates/wip-fs-server/src/main.rs @@ -0,0 +1,83 @@ +use std::error::Error; +use std::net::SocketAddr; +use std::path::PathBuf; + +use wip_fs_server::{AccessMode, ServerConfig, TransportSecurity, WipFsServer}; + +fn main() -> Result<(), Box> { + let mut root: Option = None; + let mut listen: SocketAddr = "127.0.0.1:8443".parse()?; + let mut base_path = "/wip".to_owned(); + let mut access = AccessMode::ReadOnly; + let mut security = None; + let mut arguments = std::env::args().skip(1); + while let Some(argument) = arguments.next() { + match argument.as_str() { + "--root" => root = Some(PathBuf::from(required_value(&mut arguments, "--root")?)), + "--listen" => listen = required_value(&mut arguments, "--listen")?.parse()?, + "--base-path" => base_path = required_value(&mut arguments, "--base-path")?, + "--read-write" => access = AccessMode::ReadWrite, + "--development-http" => { + set_security(&mut security, TransportSecurity::DevelopmentHttp)? + } + "--behind-tls-proxy" => { + set_security(&mut security, TransportSecurity::ReverseProxyTls)? + } + "--help" | "-h" => { + print_help(); + return Ok(()); + } + unknown => return Err(format!("unknown argument `{unknown}`; use --help").into()), + } + } + + let root = root.ok_or("--root is required")?; + let transport_security = + security.ok_or("select exactly one of --behind-tls-proxy or --development-http")?; + let mut config = ServerConfig::for_root(root); + config.listen = listen; + config.endpoint_base_path = base_path; + config.access = access; + config.transport_security = transport_security; + + let server = WipFsServer::new(config)?; + let effective_root = server.effective_root().to_owned(); + let endpoint = server.endpoint().clone(); + let bound = server.bind()?; + eprintln!( + "wip-fs-server root={} access={access:?} transport={transport_security:?} endpoint={} listen={}", + effective_root.display(), + endpoint, + bound.local_addr() + ); + bound.serve()?; + Ok(()) +} + +fn required_value( + arguments: &mut impl Iterator, + option: &str, +) -> Result> { + arguments + .next() + .ok_or_else(|| format!("{option} requires a value").into()) +} + +fn set_security( + security: &mut Option, + value: TransportSecurity, +) -> Result<(), Box> { + if security.replace(value).is_some() { + Err("select only one transport-security mode".into()) + } else { + Ok(()) + } +} + +fn print_help() { + println!( + "Usage: wip-fs-server --root PATH [--listen 127.0.0.1:8443] \\\n [--base-path /wip] [--read-write] \\\n (--behind-tls-proxy | --development-http)\n\n\ + The built-in listener is cleartext and loopback-only. In production, \ + select --behind-tls-proxy and terminate HTTPS at a trusted reverse proxy." + ); +} diff --git a/crates/wip-fs-server/src/projection.rs b/crates/wip-fs-server/src/projection.rs new file mode 100644 index 0000000..aa65395 --- /dev/null +++ b/crates/wip-fs-server/src/projection.rs @@ -0,0 +1,1034 @@ +use std::collections::VecDeque; +use std::ffi::CStr; +use std::mem::MaybeUninit; +use std::os::fd::{AsRawFd, OwnedFd}; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex}; + +use rustix::fs::{ + CWD, FileType, FlockOperation, Mode as FsMode, OFlags, RawDir, flock, fstat, ftruncate, openat, + readlinkat, +}; +use rustix::io::{dup, pread, pwrite}; +use wip_host::{ + EffectGuarantee, ExecutionContext, HostError, InterfaceRegistry, ObjectProvider, + OperationAttempt, ProviderResult, PublishedObject, RegisteredInterface, Validator, WipObject, +}; +use wip_interface as wip; +use wip_protocol::{ + FetchInterfaceRequest, FetchInterfaceResponse, FetchTreeRequest, FetchTreeResponse, Object, + ObjectTreeNode, ProtocolError, ProtocolErrorCode, +}; + +use crate::interfaces::{ + FileRead, FileReadDefinition, FileWrite, FileWriteDefinition, SymlinkInspect, + SymlinkInspectDefinition, +}; +use crate::{AccessMode, ResourceLimits}; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum EntryKind { + Directory, + Regular, + Symlink, + Unsupported, +} + +impl EntryKind { + fn from_mode(mode: u32) -> Self { + match FileType::from_raw_mode(mode) { + FileType::Directory => Self::Directory, + FileType::RegularFile => Self::Regular, + FileType::Symlink => Self::Symlink, + _ => Self::Unsupported, + } + } + + fn description(self) -> &'static str { + match self { + Self::Directory => "Filesystem directory", + Self::Regular => "Regular filesystem file", + Self::Symlink => "Symbolic link (not followed)", + Self::Unsupported => "Unsupported filesystem entry", + } + } +} + +#[derive(Debug)] +struct HandleCounter { + current: AtomicUsize, + maximum: usize, +} + +impl HandleCounter { + fn acquire(self: &Arc) -> Result { + let mut current = self.current.load(Ordering::Acquire); + loop { + if current >= self.maximum { + return Err(ProjectionError::Resource); + } + match self.current.compare_exchange_weak( + current, + current + 1, + Ordering::AcqRel, + Ordering::Acquire, + ) { + Ok(_) => { + return Ok(HandlePermit { + counter: Arc::clone(self), + }); + } + Err(actual) => current = actual, + } + } + } +} + +#[derive(Debug)] +struct HandlePermit { + counter: Arc, +} + +impl Drop for HandlePermit { + fn drop(&mut self) { + self.counter.current.fetch_sub(1, Ordering::AcqRel); + } +} + +#[derive(Debug)] +struct EntryInner { + fd: OwnedFd, + _permit: HandlePermit, +} + +#[derive(Debug, Clone)] +struct Entry { + name: String, + kind: EntryKind, + validator: Option, + initial_validator: Option>, + inner: Arc, +} + +impl Entry { + fn fd(&self) -> &OwnedFd { + &self.inner.fd + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum ProjectionError { + NotFound, + Resource, + Internal, +} + +impl ProjectionError { + fn protocol(self) -> ProtocolError { + match self { + Self::NotFound => ProtocolError { + code: ProtocolErrorCode::NotFound, + message: "target path is not published".into(), + }, + Self::Resource => ProtocolError { + code: ProtocolErrorCode::ResourceLimitExceeded, + message: "filesystem projection resource limit exceeded".into(), + }, + Self::Internal => ProtocolError { + code: ProtocolErrorCode::Internal, + message: "filesystem projection failed".into(), + }, + } + } + + fn host(self) -> ProviderResult<()> { + match self { + Self::NotFound => ProviderResult::NotFound, + Self::Resource => ProviderResult::Failure(HostError::resource_limit( + "filesystem projection resource limit exceeded", + EffectGuarantee::NotCommitted, + )), + Self::Internal => ProviderResult::Failure(HostError::internal( + "filesystem projection failed", + EffectGuarantee::NotCommitted, + )), + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum FailureInjection { + None, + BeforeWrite, + AfterTruncate, +} + +/// Request-time filesystem provider shared by the transport adapter. +pub(crate) struct Projection { + root: OwnedFd, + registry: InterfaceRegistry, + read_interface: RegisteredInterface, + symlink_interface: RegisteredInterface, + write_interface: RegisteredInterface, + access: AccessMode, + limits: ResourceLimits, + handles: Arc, + write_gate: Mutex<()>, + injection: Mutex, +} + +impl Projection { + pub(crate) fn open( + root: &std::path::Path, + access: AccessMode, + limits: ResourceLimits, + ) -> Result { + let root = openat( + CWD, + root, + OFlags::RDONLY | OFlags::DIRECTORY | OFlags::NOFOLLOW | OFlags::CLOEXEC, + FsMode::empty(), + ) + .map_err(std::io::Error::from)?; + let mut registry = InterfaceRegistry::new(); + let read_interface = registry + .register(FileReadDefinition, None) + .map_err(std::io::Error::other)?; + let symlink_interface = registry + .register(SymlinkInspectDefinition, None) + .map_err(std::io::Error::other)?; + let write_interface = registry + .register(FileWriteDefinition, None) + .map_err(std::io::Error::other)?; + Ok(Self { + root, + registry, + read_interface, + symlink_interface, + write_interface, + access, + limits, + handles: Arc::new(HandleCounter { + current: AtomicUsize::new(0), + maximum: limits.max_open_handles, + }), + write_gate: Mutex::new(()), + injection: Mutex::new(FailureInjection::None), + }) + } + + pub(crate) fn fetch_interface( + &self, + request: &FetchInterfaceRequest, + ) -> Result { + self.registry + .fetch_interface(&request.interface)? + .protocol_response() + } + + pub(crate) fn fetch_tree( + &self, + context: &ExecutionContext, + request: &FetchTreeRequest, + ) -> Result { + request.validate().map_err(|_| ProtocolError { + code: ProtocolErrorCode::InvalidRequest, + message: "invalid fetch_tree request".into(), + })?; + if request.depth > self.limits.max_tree_depth || request.limit > self.limits.max_tree_nodes + { + return Err(ProjectionError::Resource.protocol()); + } + + let root = self + .resolve(context, &request.path) + .map_err(ProjectionError::protocol)?; + let root_object = self + .publication(root.clone(), context.clone()) + .protocol_object(); + let mut objects = vec![root_object]; + let mut children: Vec> = vec![Vec::new()]; + let mut queue = VecDeque::from([(0_usize, root, 0_u32)]); + let mut truncated = false; + + while let Some((parent_index, parent, depth)) = queue.pop_front() { + check_context(context).map_err(ProjectionError::protocol)?; + if parent.kind != EntryKind::Directory { + continue; + } + let directory_fd = reopen(parent.fd(), OFlags::RDONLY | OFlags::DIRECTORY) + .map_err(ProjectionError::protocol)?; + let mut buffer = [MaybeUninit::::uninit(); 8192]; + let mut directory = RawDir::new(directory_fd, &mut buffer); + while let Some(item) = directory.next() { + check_context(context).map_err(ProjectionError::protocol)?; + let item = item.map_err(|_| ProjectionError::Internal.protocol())?; + let name = item.file_name(); + if name == cstr_dot() || name == cstr_dot_dot() { + continue; + } + let Ok(name) = name.to_str() else { + // WIP names are Unicode. Non-Unicode backend names are not published. + continue; + }; + if name.contains('/') || name.is_empty() || name == "." || name == ".." { + continue; + } + let child = match self.open_child(context, parent.fd(), name) { + Ok(child) => child, + Err(ProjectionError::NotFound) => continue, + Err(error) => return Err(error.protocol()), + }; + if depth == request.depth || objects.len() == request.limit as usize { + truncated = true; + break; + } + let child_index = objects.len(); + objects.push( + self.publication(child.clone(), context.clone()) + .protocol_object(), + ); + children.push(Vec::new()); + children[parent_index].push(child_index); + if child.kind == EntryKind::Directory { + queue.push_back((child_index, child, depth + 1)); + } + } + if truncated && objects.len() == request.limit as usize { + break; + } + } + + fn build(index: usize, objects: &[Object], children: &[Vec]) -> ObjectTreeNode { + ObjectTreeNode { + object: objects[index].clone(), + children: children[index] + .iter() + .map(|child| build(*child, objects, children)) + .collect(), + } + } + + let response = FetchTreeResponse { + root: build(0, &objects, &children), + truncated, + }; + response.validate_for(request).map_err(|_| ProtocolError { + code: ProtocolErrorCode::Internal, + message: "Host produced an invalid tree response".into(), + })?; + Ok(response) + } + + fn publication( + &self, + entry: Entry, + context: ExecutionContext, + ) -> PublishedObject { + let kind = entry.kind; + let mut publication = PublishedObject::new(FilesystemObject { + entry, + context, + limits: self.limits, + injection: *self + .injection + .lock() + .unwrap_or_else(|error| error.into_inner()), + }); + match kind { + EntryKind::Regular => { + publication = publication + .expose(&self.read_interface) + .expect("regular files implement FileRead"); + if self.access == AccessMode::ReadWrite { + publication = publication + .expose(&self.write_interface) + .expect("regular files implement FileWrite"); + } + } + EntryKind::Symlink => { + publication = publication + .expose(&self.symlink_interface) + .expect("symbolic links implement SymlinkInspect"); + } + EntryKind::Directory | EntryKind::Unsupported => {} + } + publication + } + + fn resolve(&self, context: &ExecutionContext, path: &str) -> Result { + check_context(context)?; + wip_protocol::validate_path(path).map_err(|_| ProjectionError::NotFound)?; + if path == "/" { + let fd = reopen(&self.root, OFlags::RDONLY | OFlags::DIRECTORY)?; + return self.entry(context, String::new(), fd); + } + let mut parent = dup(&self.root).map_err(|_| ProjectionError::Internal)?; + let mut segments = path[1..].split('/').peekable(); + while let Some(segment) = segments.next() { + check_context(context)?; + let probe = openat( + &parent, + segment, + OFlags::PATH | OFlags::NOFOLLOW | OFlags::CLOEXEC, + FsMode::empty(), + ) + .map_err(classify_lookup)?; + let stat = fstat(&probe).map_err(|_| ProjectionError::Internal)?; + let kind = EntryKind::from_mode(stat.st_mode); + if segments.peek().is_some() { + if kind != EntryKind::Directory { + return Err(ProjectionError::NotFound); + } + parent = reopen(&probe, OFlags::RDONLY | OFlags::DIRECTORY)?; + } else { + let fd = self.reopen_final(probe, kind)?; + return self.entry(context, segment.to_owned(), fd); + } + } + Err(ProjectionError::NotFound) + } + + fn open_child( + &self, + context: &ExecutionContext, + parent: &OwnedFd, + name: &str, + ) -> Result { + check_context(context)?; + let probe = openat( + parent, + name, + OFlags::PATH | OFlags::NOFOLLOW | OFlags::CLOEXEC, + FsMode::empty(), + ) + .map_err(classify_lookup)?; + let stat = fstat(&probe).map_err(|_| ProjectionError::Internal)?; + let kind = EntryKind::from_mode(stat.st_mode); + let fd = self.reopen_final(probe, kind)?; + self.entry(context, name.to_owned(), fd) + } + + fn reopen_final(&self, probe: OwnedFd, kind: EntryKind) -> Result { + match kind { + EntryKind::Directory => reopen(&probe, OFlags::RDONLY | OFlags::DIRECTORY), + EntryKind::Regular => { + let access = if self.access == AccessMode::ReadWrite { + OFlags::RDWR + } else { + OFlags::RDONLY + }; + reopen(&probe, access) + } + EntryKind::Symlink | EntryKind::Unsupported => Ok(probe), + } + } + + fn entry( + &self, + context: &ExecutionContext, + name: String, + fd: OwnedFd, + ) -> Result { + check_context(context)?; + let permit = self.handles.acquire()?; + let stat = fstat(&fd).map_err(|_| ProjectionError::Internal)?; + let kind = EntryKind::from_mode(stat.st_mode); + let validator = matches!(kind, EntryKind::Regular | EntryKind::Symlink) + .then(|| validator_from_stat(&stat)); + Ok(Entry { + name, + kind, + initial_validator: validator.clone(), + validator: validator.map(Validator::new), + inner: Arc::new(EntryInner { + fd, + _permit: permit, + }), + }) + } + + #[cfg(test)] + fn set_injection(&self, injection: FailureInjection) { + *self + .injection + .lock() + .unwrap_or_else(|error| error.into_inner()) = injection; + } +} + +impl ObjectProvider for Projection { + type Object = FilesystemObject; + + fn with_object( + &self, + context: &ExecutionContext, + path: &str, + operation: impl FnOnce(&PublishedObject) -> R, + ) -> ProviderResult { + let entry = match self.resolve(context, path) { + Ok(entry) => entry, + Err(error) => { + return match error.host() { + ProviderResult::NotFound => ProviderResult::NotFound, + ProviderResult::Failure(error) => ProviderResult::Failure(error), + ProviderResult::Found(()) => unreachable!(), + }; + } + }; + let publication = self.publication(entry, context.clone()); + ProviderResult::Found(operation(&publication)) + } + + fn execute_operation( + &self, + context: &ExecutionContext, + path: &str, + operation: impl FnOnce(&PublishedObject) -> OperationAttempt, + ) -> ProviderResult { + let _gate = match self.write_gate.lock() { + Ok(gate) => gate, + Err(_) => { + return ProviderResult::Failure(HostError::internal( + "filesystem operation lock failed", + EffectGuarantee::NotCommitted, + )); + } + }; + let entry = match self.resolve(context, path) { + Ok(entry) => entry, + Err(ProjectionError::NotFound) => return ProviderResult::NotFound, + Err(ProjectionError::Resource) => { + return ProviderResult::Failure(HostError::resource_limit( + "filesystem projection resource limit exceeded", + EffectGuarantee::NotCommitted, + )); + } + Err(ProjectionError::Internal) => { + return ProviderResult::Failure(HostError::internal( + "filesystem projection failed", + EffectGuarantee::NotCommitted, + )); + } + }; + if entry.kind == EntryKind::Regular + && flock(entry.fd(), FlockOperation::LockExclusive).is_err() + { + return ProviderResult::Failure(HostError::internal( + "filesystem object could not be locked", + EffectGuarantee::NotCommitted, + )); + } + let publication = self.publication(entry.clone(), context.clone()); + let mut attempt = operation(&publication); + if let OperationAttempt::Started(Ok(result)) = &mut attempt { + let stat = match fstat(entry.fd()) { + Ok(stat) => stat, + Err(_) => { + return ProviderResult::Failure(HostError::internal( + "filesystem state could not be observed after dispatch", + EffectGuarantee::Unknown, + )); + } + }; + result.validator = matches!(entry.kind, EntryKind::Regular | EntryKind::Symlink) + .then(|| validator_from_stat(&stat)); + } + if entry.kind == EntryKind::Regular { + let _ = flock(entry.fd(), FlockOperation::Unlock); + } + ProviderResult::Found(attempt) + } +} + +pub(crate) struct FilesystemObject { + entry: Entry, + context: ExecutionContext, + limits: ResourceLimits, + injection: FailureInjection, +} + +impl WipObject for FilesystemObject { + fn name(&self) -> &str { + &self.entry.name + } + + fn description(&self) -> Option<&str> { + Some(self.entry.kind.description()) + } + + fn validator(&self) -> Option<&Validator> { + self.entry.validator.as_ref() + } + + fn requires_validator(&self, _: &wip_host::InterfaceRef, operation: &str) -> bool { + operation == "replace" + } +} + +impl FileRead for FilesystemObject { + fn read(&self, offset: Option, length: Option) -> wip::OperationResult { + check_context(&self.context).map_err(resource_host)?; + let offset = offset.unwrap_or(0); + let stat = fstat(self.entry.fd()).map_err(|_| internal_not_committed())?; + let size = u64::try_from(stat.st_size).map_err(|_| internal_not_committed())?; + let available = size.saturating_sub(offset); + let length = match length { + Some(length) if length <= self.limits.max_file_read_bytes as u64 => { + length.min(available) + } + Some(_) => return Err(resource_not_committed()), + None if available <= self.limits.max_file_read_bytes as u64 => available, + None => return Err(resource_not_committed()), + }; + let length = usize::try_from(length).map_err(|_| resource_not_committed())?; + let mut bytes = vec![0_u8; length]; + let mut read = 0_usize; + while read < length { + check_context(&self.context).map_err(resource_host)?; + let count = pread(self.entry.fd(), &mut bytes[read..], offset + read as u64) + .map_err(|_| internal_not_committed())?; + if count == 0 { + break; + } + read += count; + } + bytes.truncate(read); + Ok(wip::Bytes(bytes)) + } +} + +impl SymlinkInspect for FilesystemObject { + fn read_link(&self) -> wip::OperationResult { + check_context(&self.context).map_err(resource_host)?; + let target = + readlinkat(self.entry.fd(), "", Vec::new()).map_err(|_| internal_not_committed())?; + Ok(wip::Bytes(target.as_bytes().to_vec())) + } +} + +impl FileWrite for FilesystemObject { + fn replace(&self, contents: wip::Bytes) -> wip::OperationResult<()> { + check_context(&self.context).map_err(resource_host)?; + if contents.0.len() > self.limits.max_file_write_bytes { + return Err(resource_not_committed()); + } + let current = fstat(self.entry.fd()).map_err(|_| internal_not_committed())?; + if self.entry.initial_validator.as_deref() != Some(validator_from_stat(¤t).as_slice()) + { + return Err(HostError::internal( + "filesystem object changed before write", + EffectGuarantee::NotCommitted, + )); + } + if self.injection == FailureInjection::BeforeWrite { + return Err(internal_not_committed()); + } + ftruncate(self.entry.fd(), 0).map_err(|_| internal_not_committed())?; + if self.injection == FailureInjection::AfterTruncate { + return Err(HostError::internal( + "injected failure after write began", + EffectGuarantee::Unknown, + )); + } + let mut written = 0_usize; + while written < contents.0.len() { + if context_expired(&self.context) { + return Err(HostError::resource_limit( + "write interrupted after execution began", + EffectGuarantee::Unknown, + )); + } + let count = + pwrite(self.entry.fd(), &contents.0[written..], written as u64).map_err(|_| { + HostError::internal("filesystem write failed", EffectGuarantee::Unknown) + })?; + if count == 0 { + return Err(HostError::internal( + "filesystem write made no progress", + EffectGuarantee::Unknown, + )); + } + written += count; + } + rustix::fs::fdatasync(self.entry.fd()).map_err(|_| { + HostError::internal( + "filesystem synchronization failed", + EffectGuarantee::Unknown, + ) + })?; + Ok(()) + } +} + +fn reopen(fd: &OwnedFd, flags: OFlags) -> Result { + let path = format!("/proc/self/fd/{}", fd.as_raw_fd()); + openat(CWD, path, flags | OFlags::CLOEXEC, FsMode::empty()).map_err(classify_lookup) +} + +fn validator_from_stat(stat: &rustix::fs::Stat) -> Vec { + let mut bytes = Vec::with_capacity(56); + bytes.extend_from_slice(&stat.st_dev.to_be_bytes()); + bytes.extend_from_slice(&stat.st_ino.to_be_bytes()); + bytes.extend_from_slice(&u64::from(stat.st_mode).to_be_bytes()); + bytes.extend_from_slice(&stat.st_size.to_be_bytes()); + bytes.extend_from_slice(&stat.st_mtime.to_be_bytes()); + bytes.extend_from_slice(&stat.st_mtime_nsec.to_be_bytes()); + bytes.extend_from_slice(&stat.st_ctime.to_be_bytes()); + bytes.extend_from_slice(&stat.st_ctime_nsec.to_be_bytes()); + bytes +} + +fn classify_lookup(error: rustix::io::Errno) -> ProjectionError { + use rustix::io::Errno; + if matches!( + error, + Errno::NOENT | Errno::NOTDIR | Errno::LOOP | Errno::ACCESS | Errno::PERM + ) { + ProjectionError::NotFound + } else { + ProjectionError::Internal + } +} + +fn context_expired(context: &ExecutionContext) -> bool { + context.cancellation().is_cancelled() + || context + .deadline() + .is_some_and(|deadline| std::time::Instant::now() >= deadline) +} + +fn check_context(context: &ExecutionContext) -> Result<(), ProjectionError> { + if context_expired(context) { + Err(ProjectionError::Resource) + } else { + Ok(()) + } +} + +fn resource_host(_: ProjectionError) -> HostError { + resource_not_committed() +} + +fn resource_not_committed() -> HostError { + HostError::resource_limit( + "filesystem operation resource limit exceeded", + EffectGuarantee::NotCommitted, + ) +} + +fn internal_not_committed() -> HostError { + HostError::internal("filesystem operation failed", EffectGuarantee::NotCommitted) +} + +fn cstr_dot() -> &'static CStr { + c"." +} + +fn cstr_dot_dot() -> &'static CStr { + c".." +} + +#[cfg(test)] +mod tests { + use std::collections::BTreeMap; + use std::fs; + + use tempfile::tempdir; + use wip_host::{ + CallErrorKind, CallRequest, ExecutionContext, InterfaceTarget, ObjectProvider, Target, + }; + use wip_protocol::{FetchTreeRequest, ProtocolErrorCode, Value}; + + use super::{FailureInjection, Projection}; + use crate::{AccessMode, ResourceLimits}; + + fn projection(root: &std::path::Path, access: AccessMode) -> Projection { + Projection::open(root, access, ResourceLimits::default()).expect("open projection") + } + + fn call( + projection: &Projection, + path: &str, + validator: Option>, + interface: String, + operation: &str, + arguments: BTreeMap, + ) -> Result { + projection.call_operation(CallRequest { + target: Target { + path: path.into(), + validator, + }, + interface: InterfaceTarget { + reference: interface, + validator: None, + }, + operation: operation.into(), + arguments, + }) + } + + #[test] + fn projects_entry_kinds_without_following_symlinks() { + let temporary = tempdir().expect("temporary root"); + fs::write(temporary.path().join("file.txt"), b"hello").expect("write file"); + fs::create_dir(temporary.path().join("dir")).expect("create directory"); + #[cfg(unix)] + std::os::unix::fs::symlink("/etc/passwd", temporary.path().join("link")) + .expect("create symlink"); + let projection = projection(temporary.path(), AccessMode::ReadOnly); + + let root = projection.fetch("/").expect("fetch root"); + assert_eq!(root.name, ""); + assert_eq!(root.description.as_deref(), Some("Filesystem directory")); + assert!(root.interfaces.is_empty()); + assert!(root.validator.is_none()); + + let file = projection.fetch("/file.txt").expect("fetch file"); + assert_eq!(file.interfaces.len(), 1); + assert!(file.validator.is_some()); + assert!(file.r#ref.is_none()); + + let directory = projection.fetch("/dir").expect("fetch directory"); + assert!(directory.interfaces.is_empty()); + + #[cfg(unix)] + { + let link = projection.fetch("/link").expect("fetch link itself"); + assert_eq!( + link.description.as_deref(), + Some("Symbolic link (not followed)") + ); + assert_eq!(link.interfaces.len(), 1); + assert_eq!( + projection + .fetch("/link/shadow") + .expect_err("must not follow") + .kind(), + CallErrorKind::NotFound + ); + let target = call( + &projection, + "/link", + link.validator, + link.interfaces[0].clone(), + "read_link", + BTreeMap::new(), + ) + .expect("inspect symlink"); + assert_eq!(target.result, Value::Bytes(b"/etc/passwd".to_vec())); + } + + assert_eq!( + projection.fetch("/missing").expect_err("missing").kind(), + CallErrorKind::NotFound + ); + } + + #[test] + fn reads_are_bounded_and_tree_truncation_is_exact() { + let temporary = tempdir().expect("temporary root"); + fs::write(temporary.path().join("a"), b"abcdef").expect("write file"); + fs::write(temporary.path().join("b"), b"b").expect("write file"); + let projection = projection(temporary.path(), AccessMode::ReadOnly); + let file = projection.fetch("/a").expect("fetch file"); + let result = call( + &projection, + "/a", + file.validator.clone(), + file.interfaces[0].clone(), + "read", + BTreeMap::from([ + ("offset".into(), Value::Integer(2)), + ("length".into(), Value::Integer(3)), + ]), + ) + .expect("read range"); + assert_eq!(result.result, Value::Bytes(b"cde".to_vec())); + + let complete = projection + .fetch_tree( + &ExecutionContext::new(), + &FetchTreeRequest { + path: "/".into(), + depth: 1, + limit: 3, + }, + ) + .expect("complete tree"); + assert!(!complete.truncated); + assert_eq!(complete.root.children.len(), 2); + + let limited = projection + .fetch_tree( + &ExecutionContext::new(), + &FetchTreeRequest { + path: "/".into(), + depth: 1, + limit: 2, + }, + ) + .expect("limited tree"); + assert!(limited.truncated); + assert_eq!(limited.root.children.len(), 1); + } + + #[test] + fn writes_require_current_validator_and_return_new_validator() { + let temporary = tempdir().expect("temporary root"); + let path = temporary.path().join("file"); + fs::write(&path, b"old").expect("write file"); + let projection = projection(temporary.path(), AccessMode::ReadWrite); + let observed = projection.fetch("/file").expect("fetch file"); + assert_eq!(observed.interfaces.len(), 2); + let write_interface = observed.interfaces[1].clone(); + + let missing = call( + &projection, + "/file", + None, + write_interface.clone(), + "replace", + BTreeMap::from([("contents".into(), Value::Bytes(b"new".to_vec()))]), + ) + .expect_err("write validator required"); + assert_eq!(missing.kind(), CallErrorKind::ValidatorRequired); + + let stale_validator = observed.validator.clone(); + fs::remove_file(&path).expect("remove old entry"); + fs::write(&path, b"replacement").expect("replace path"); + let stale = call( + &projection, + "/file", + stale_validator, + write_interface, + "replace", + BTreeMap::from([("contents".into(), Value::Bytes(b"bad".to_vec()))]), + ) + .expect_err("replacement identity must mismatch"); + assert_eq!(stale.kind(), CallErrorKind::ValidatorMismatch); + assert_eq!(fs::read(&path).expect("read replacement"), b"replacement"); + + let current = projection.fetch("/file").expect("fetch replacement"); + let result = call( + &projection, + "/file", + current.validator.clone(), + current.interfaces[1].clone(), + "replace", + BTreeMap::from([("contents".into(), Value::Bytes(b"committed".to_vec()))]), + ) + .expect("replace file"); + assert_ne!(result.validator, current.validator); + assert_eq!(fs::read(path).expect("read committed file"), b"committed"); + } + + #[test] + fn post_dispatch_failures_preserve_effect_guarantees() { + let temporary = tempdir().expect("temporary root"); + let path = temporary.path().join("file"); + fs::write(&path, b"old").expect("write file"); + let projection = projection(temporary.path(), AccessMode::ReadWrite); + let observed = projection.fetch("/file").expect("fetch file"); + let request = || { + call( + &projection, + "/file", + observed.validator.clone(), + observed.interfaces[1].clone(), + "replace", + BTreeMap::from([("contents".into(), Value::Bytes(b"new".to_vec()))]), + ) + }; + + projection.set_injection(FailureInjection::BeforeWrite); + assert_eq!( + request().expect_err("pre-write failure").kind(), + CallErrorKind::Internal + ); + assert_eq!(fs::read(&path).expect("old file remains"), b"old"); + + projection.set_injection(FailureInjection::AfterTruncate); + assert_eq!( + request().expect_err("post-truncate failure").kind(), + CallErrorKind::OperationOutcomeUnknown + ); + assert_eq!(fs::read(path).expect("truncated file"), b""); + } + + #[test] + fn large_directory_stops_after_proving_truncation() { + let temporary = tempdir().expect("temporary root"); + for index in 0..2_000 { + fs::write(temporary.path().join(format!("entry-{index}")), b"x") + .expect("write directory entry"); + } + let projection = projection(temporary.path(), AccessMode::ReadOnly); + let tree = projection + .fetch_tree( + &ExecutionContext::new(), + &FetchTreeRequest { + path: "/".into(), + depth: 1, + limit: 2, + }, + ) + .expect("bounded tree"); + assert!(tree.truncated); + assert_eq!(tree.root.children.len(), 1); + } + + #[cfg(unix)] + #[test] + fn non_unicode_names_are_hidden_and_special_files_are_explicit() { + use std::ffi::OsString; + use std::os::unix::ffi::OsStringExt; + use std::os::unix::net::UnixListener; + + let temporary = tempdir().expect("temporary root"); + let invalid_name = OsString::from_vec(vec![b'n', b'o', b'n', 0xff]); + fs::write(temporary.path().join(invalid_name), b"hidden").expect("write non-Unicode file"); + let socket_path = temporary.path().join("socket"); + let _socket = UnixListener::bind(&socket_path).expect("bind filesystem socket"); + let projection = projection(temporary.path(), AccessMode::ReadOnly); + + let tree = projection + .fetch_tree( + &ExecutionContext::new(), + &FetchTreeRequest { + path: "/".into(), + depth: 1, + limit: 10, + }, + ) + .expect("list root"); + assert_eq!(tree.root.children.len(), 1); + assert_eq!(tree.root.children[0].object.name, "socket"); + assert_eq!( + tree.root.children[0].object.description.as_deref(), + Some("Unsupported filesystem entry") + ); + assert!(tree.root.children[0].object.interfaces.is_empty()); + } + + #[test] + fn cancellation_is_not_clean_tree_exhaustion() { + let temporary = tempdir().expect("temporary root"); + fs::write(temporary.path().join("entry"), b"x").expect("write file"); + let projection = projection(temporary.path(), AccessMode::ReadOnly); + let cancellation = wip_host::CancellationToken::new(); + cancellation.cancel(); + let error = projection + .fetch_tree( + &ExecutionContext::new().with_cancellation(cancellation), + &FetchTreeRequest { + path: "/".into(), + depth: 1, + limit: 2, + }, + ) + .expect_err("cancelled traversal"); + assert_eq!(error.code, ProtocolErrorCode::ResourceLimitExceeded); + } +} diff --git a/crates/wip-fs-server/src/server.rs b/crates/wip-fs-server/src/server.rs new file mode 100644 index 0000000..a5aa5e9 --- /dev/null +++ b/crates/wip-fs-server/src/server.rs @@ -0,0 +1,523 @@ +use std::io::Read; +use std::net::SocketAddr; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::thread; +use std::time::{Duration, Instant}; + +use http::{Request, Response, StatusCode}; +use wip_host::{ExecutionContext, ObjectProvider}; +use wip_https::{ + BatchRequestKind, BatchResponse, BatchResponseItem, BatchSuccess, CodecError, EncodeHttpError, + Endpoint, RequestMetadataError, ResourceLimitKind, Route, decode_batch_request, + decode_call_operation_metadata, decode_fetch_interface_request, decode_fetch_request, + decode_fetch_tree_request, encode_batch_error_response, + encode_batch_resource_limit_error_response, encode_batch_response, + encode_call_operation_response, encode_fetch_interface_response, encode_fetch_response, + encode_fetch_tree_response, encode_protocol_error_response, + encode_resource_limit_error_response, validate_http_request, +}; +use wip_protocol::{ProtocolError, ProtocolErrorCode, ProtocolInteraction}; + +use crate::projection::Projection; +use crate::{ConfigError, ServerConfig}; + +/// Filesystem projection and framework-neutral standard WIP request adapter. +pub struct WipFsServer { + config: ServerConfig, + endpoint: Endpoint, + effective_root: std::path::PathBuf, + projection: Arc, +} + +impl WipFsServer { + /// Validates configuration and opens the configured directory capability. + pub fn new(config: ServerConfig) -> Result { + config.validate()?; + let effective_root = config.effective_root()?; + let endpoint = config.endpoint()?; + let projection = Projection::open(&effective_root, config.access, config.limits)?; + Ok(Self { + config, + endpoint, + effective_root, + projection: Arc::new(projection), + }) + } + + /// Returns the canonical effective filesystem root opened by the server. + #[must_use] + pub fn effective_root(&self) -> &std::path::Path { + &self.effective_root + } + + /// Returns the configured transport endpoint prefix. + #[must_use] + pub fn endpoint(&self) -> &Endpoint { + &self.endpoint + } + + /// Handles one already-buffered HTTP request with the standard `wip-https` codecs. + /// + /// Listener integrations must enforce the same body limit while collecting bytes. + #[must_use] + pub fn handle_http(&self, request: Request>) -> Response> { + let Some(route) = self.route(request.uri().path()) else { + return empty_response(StatusCode::NOT_FOUND); + }; + if let Err(error) = validate_http_request(&self.endpoint, route, &request) { + return self.metadata_error(route, error); + } + let context = ExecutionContext::new() + .with_deadline(Instant::now() + self.config.limits.request_timeout) + .with_resource_budget(u64::from(self.config.limits.max_tree_nodes)); + match route { + Route::Fetch => self.fetch(request.body(), &context), + Route::FetchTree => self.fetch_tree(request.body(), &context), + Route::FetchInterface => self.fetch_interface(request.body()), + Route::CallOperation => self.call_operation(request.body(), &context), + Route::Batch => self.batch(request.body(), &context), + } + } + + /// Binds the built-in bounded loopback HTTP listener. + /// + /// The listener is cleartext by design. Configuration must explicitly state + /// whether TLS is terminated by a reverse proxy or development HTTP is in use. + pub fn bind(self) -> Result { + let server = tiny_http::Server::http(self.config.listen) + .map_err(|error| ServerBuildError::Listener(error.to_string()))?; + let local_addr = server + .server_addr() + .to_ip() + .ok_or_else(|| ServerBuildError::Listener("expected an IP listener".into()))?; + Ok(BoundServer { + local_addr, + service: Arc::new(self), + listener: Arc::new(server), + }) + } + + fn route(&self, path: &str) -> Option { + [ + Route::Fetch, + Route::FetchTree, + Route::FetchInterface, + Route::CallOperation, + Route::Batch, + ] + .into_iter() + .find(|route| self.endpoint.route(*route).path() == path) + } + + fn fetch(&self, body: &[u8], context: &ExecutionContext) -> Response> { + let request = match decode_fetch_request(body, self.config.limits.wire) { + Ok(request) => request, + Err(error) => return self.codec_error(ProtocolInteraction::Fetch, error), + }; + match self + .projection + .fetch_with_context(context, &request.path) + .map_err(|error| error.protocol_error()) + { + Ok(object) => encode_fetch_response(&request, &object, self.config.limits.wire) + .unwrap_or_else(|error| self.read_encode_error(ProtocolInteraction::Fetch, error)), + Err(error) => self.protocol_error(ProtocolInteraction::Fetch, &error, None), + } + } + + fn fetch_tree(&self, body: &[u8], context: &ExecutionContext) -> Response> { + let request = match decode_fetch_tree_request(body, self.config.limits.wire) { + Ok(request) => request, + Err(error) => return self.codec_error(ProtocolInteraction::FetchTree, error), + }; + match self.projection.fetch_tree(context, &request) { + Ok(tree) => encode_fetch_tree_response(&request, &tree, self.config.limits.wire) + .unwrap_or_else(|error| { + self.read_encode_error(ProtocolInteraction::FetchTree, error) + }), + Err(error) => self.protocol_error(ProtocolInteraction::FetchTree, &error, None), + } + } + + fn fetch_interface(&self, body: &[u8]) -> Response> { + let request = match decode_fetch_interface_request(body, self.config.limits.wire) { + Ok(request) => request, + Err(error) => return self.codec_error(ProtocolInteraction::FetchInterface, error), + }; + match self.projection.fetch_interface(&request) { + Ok(interface) => { + encode_fetch_interface_response(&request, &interface, self.config.limits.wire) + .unwrap_or_else(|error| { + self.read_encode_error(ProtocolInteraction::FetchInterface, error) + }) + } + Err(error) => self.protocol_error(ProtocolInteraction::FetchInterface, &error, None), + } + } + + fn call_operation(&self, body: &[u8], context: &ExecutionContext) -> Response> { + let metadata = match decode_call_operation_metadata(body, self.config.limits.wire) { + Ok(metadata) => metadata, + Err(error) => return self.codec_error(ProtocolInteraction::CallOperation, error), + }; + let decoded = std::sync::Mutex::new(None); + let execution = self.projection.call_operation_with_context_and_decoder( + context, + metadata.target.clone(), + metadata.interface.clone(), + metadata.operation.clone(), + |descriptor| { + let request = metadata.decode_request(body, descriptor, self.config.limits.wire)?; + let arguments = request.arguments.clone(); + *decoded.lock().unwrap_or_else(|error| error.into_inner()) = + Some((request, descriptor.clone())); + Ok::<_, CodecError>(arguments) + }, + ); + match execution.map_err(|error| error.protocol_error()) { + Ok(result) => { + let Some((request, descriptor)) = decoded + .into_inner() + .unwrap_or_else(|error| error.into_inner()) + else { + return self.protocol_error( + ProtocolInteraction::CallOperation, + &ProtocolError { + code: ProtocolErrorCode::OperationOutcomeUnknown, + message: "operation completed without response metadata".into(), + }, + None, + ); + }; + encode_call_operation_response( + &request, + &descriptor, + &result, + self.config.limits.wire, + ) + .unwrap_or_else(|_| { + // Dispatch succeeded. A transport/result encoding failure cannot + // prove that a write did not commit. + let error = ProtocolError { + code: ProtocolErrorCode::OperationOutcomeUnknown, + message: "operation completed but its response could not be encoded".into(), + }; + self.protocol_error(ProtocolInteraction::CallOperation, &error, None) + }) + } + Err(error) => self.protocol_error(ProtocolInteraction::CallOperation, &error, None), + } + } + + fn batch(&self, body: &[u8], context: &ExecutionContext) -> Response> { + let request = match decode_batch_request(body, self.config.limits.wire) { + Ok(request) => request, + Err(error) => return self.batch_codec_error(error), + }; + let responses = request + .requests + .iter() + .map(|item| { + let outcome = match &item.request { + BatchRequestKind::Fetch(request) => self + .projection + .fetch_with_context(context, &request.path) + .map(BatchSuccess::Fetch) + .map_err(|error| error.protocol_error()), + BatchRequestKind::FetchTree(request) => self + .projection + .fetch_tree(context, request) + .map(BatchSuccess::FetchTree), + BatchRequestKind::FetchInterface(request) => self + .projection + .fetch_interface(request) + .map(BatchSuccess::FetchInterface), + BatchRequestKind::Invalid { error, .. } => Err(error.clone()), + }; + BatchResponseItem { + id: item.id.clone(), + outcome, + } + }) + .collect(); + let response = BatchResponse { responses }; + encode_batch_response(&request, &response, self.config.limits.wire).unwrap_or_else( + |error| match error { + EncodeHttpError::Codec(codec) => self.batch_codec_error(codec), + EncodeHttpError::Http(_) => self.batch_top_error(&internal_error(), None), + }, + ) + } + + fn metadata_error(&self, route: Route, error: RequestMetadataError) -> Response> { + match error { + RequestMetadataError::InvalidRequest(_) => self.protocol_error( + interaction(route), + &ProtocolError { + code: ProtocolErrorCode::InvalidRequest, + message: "invalid HTTP request metadata".into(), + }, + None, + ), + RequestMetadataError::UnsupportedMediaType => { + empty_response(StatusCode::UNSUPPORTED_MEDIA_TYPE) + } + RequestMetadataError::NotAcceptable => empty_response(StatusCode::NOT_ACCEPTABLE), + } + } + + fn codec_error( + &self, + interaction: ProtocolInteraction, + error: CodecError, + ) -> Response> { + if let Some(kind) = error.resource_limit_kind() { + self.protocol_error(interaction, &resource_error(), Some(kind)) + } else { + self.protocol_error( + interaction, + &ProtocolError { + code: ProtocolErrorCode::InvalidRequest, + message: "malformed WIP request".into(), + }, + None, + ) + } + } + + fn batch_codec_error(&self, error: CodecError) -> Response> { + if let Some(kind) = error.resource_limit_kind() { + self.batch_top_error(&resource_error(), Some(kind)) + } else { + self.batch_top_error( + &ProtocolError { + code: ProtocolErrorCode::InvalidRequest, + message: "malformed WIP batch request".into(), + }, + None, + ) + } + } + + fn read_encode_error( + &self, + interaction: ProtocolInteraction, + error: EncodeHttpError, + ) -> Response> { + match error { + EncodeHttpError::Codec(codec) if codec.resource_limit_kind().is_some() => { + self.protocol_error(interaction, &resource_error(), codec.resource_limit_kind()) + } + _ => self.protocol_error(interaction, &internal_error(), None), + } + } + + fn protocol_error( + &self, + interaction: ProtocolInteraction, + error: &ProtocolError, + resource_kind: Option, + ) -> Response> { + let encoded = match resource_kind { + Some(kind) => encode_resource_limit_error_response( + interaction, + error, + kind, + self.config.limits.wire, + ), + None => encode_protocol_error_response(interaction, error, self.config.limits.wire), + }; + encoded.unwrap_or_else(|_| empty_response(StatusCode::INTERNAL_SERVER_ERROR)) + } + + fn batch_top_error( + &self, + error: &ProtocolError, + resource_kind: Option, + ) -> Response> { + let encoded = match resource_kind { + Some(kind) => { + encode_batch_resource_limit_error_response(error, kind, self.config.limits.wire) + } + None => encode_batch_error_response(error, self.config.limits.wire), + }; + encoded.unwrap_or_else(|_| empty_response(StatusCode::INTERNAL_SERVER_ERROR)) + } +} + +/// Bound real HTTP listener for a [`WipFsServer`]. +pub struct BoundServer { + local_addr: SocketAddr, + service: Arc, + listener: Arc, +} + +impl BoundServer { + /// Returns the actual bound loopback address, including an assigned port. + #[must_use] + pub fn local_addr(&self) -> SocketAddr { + self.local_addr + } + + /// Serves requests with a fixed worker count until `shutdown` becomes true. + pub fn serve_until(&self, shutdown: Arc) -> Result<(), std::io::Error> { + let mut workers = Vec::with_capacity(self.service.config.limits.max_concurrent_requests); + for _ in 0..self.service.config.limits.max_concurrent_requests { + let listener = Arc::clone(&self.listener); + let service = Arc::clone(&self.service); + let shutdown = Arc::clone(&shutdown); + workers.push(thread::spawn(move || { + while !shutdown.load(Ordering::Acquire) { + match listener.recv_timeout(Duration::from_millis(25)) { + Ok(Some(request)) => serve_request(&service, request), + Ok(None) => {} + Err(error) => return Err(error), + } + } + Ok(()) + })); + } + for worker in workers { + worker + .join() + .map_err(|_| std::io::Error::other("HTTP worker panicked"))??; + } + Ok(()) + } + + /// Serves until the process is terminated. + pub fn serve(&self) -> Result<(), std::io::Error> { + self.serve_until(Arc::new(AtomicBool::new(false))) + } +} + +fn serve_request(service: &WipFsServer, mut request: tiny_http::Request) { + let maximum = service.config.limits.wire.max_request_bytes(); + let mut body = Vec::new(); + let read_result = request + .as_reader() + .take(maximum.saturating_add(1) as u64) + .read_to_end(&mut body); + let route = service.route(request.url().split('?').next().unwrap_or_default()); + let response = if read_result.is_err() { + empty_response(StatusCode::BAD_REQUEST) + } else if body.len() > maximum { + route.map_or_else( + || empty_response(StatusCode::PAYLOAD_TOO_LARGE), + |route| { + if route == Route::Batch { + service.batch_top_error( + &resource_error(), + Some(ResourceLimitKind::RepresentationBytes), + ) + } else { + service.protocol_error( + interaction(route), + &resource_error(), + Some(ResourceLimitKind::RepresentationBytes), + ) + } + }, + ) + } else { + let mut builder = Request::builder() + .method(request.method().as_str()) + .uri(request.url()); + for header in request.headers() { + if let (Ok(name), Ok(value)) = ( + http::header::HeaderName::from_bytes(header.field.as_str().as_bytes()), + http::HeaderValue::from_bytes(header.value.as_bytes()), + ) { + builder = builder.header(name, value); + } + } + match builder.body(body) { + Ok(request) => service.handle_http(request), + Err(_) => empty_response(StatusCode::BAD_REQUEST), + } + }; + let status = tiny_http::StatusCode(response.status().as_u16()); + let mut outgoing = + tiny_http::Response::from_data(response.body().clone()).with_status_code(status); + for (name, value) in response.headers() { + if let Ok(header) = tiny_http::Header::from_bytes(name.as_str(), value.as_bytes()) { + outgoing.add_header(header); + } + } + let _ = request.respond(outgoing); +} + +fn interaction(route: Route) -> ProtocolInteraction { + match route { + Route::Fetch => ProtocolInteraction::Fetch, + Route::FetchTree => ProtocolInteraction::FetchTree, + Route::FetchInterface => ProtocolInteraction::FetchInterface, + Route::CallOperation => ProtocolInteraction::CallOperation, + Route::Batch => ProtocolInteraction::Fetch, + } +} + +fn resource_error() -> ProtocolError { + ProtocolError { + code: ProtocolErrorCode::ResourceLimitExceeded, + message: "configured resource limit exceeded".into(), + } +} + +fn internal_error() -> ProtocolError { + ProtocolError { + code: ProtocolErrorCode::Internal, + message: "filesystem server failed".into(), + } +} + +fn empty_response(status: StatusCode) -> Response> { + Response::builder() + .status(status) + .body(Vec::new()) + .expect("static status builds an HTTP response") +} + +/// Failure while validating, opening, or binding the server. +#[derive(Debug)] +pub enum ServerBuildError { + /// Configuration is unsafe or unbounded. + Config(ConfigError), + /// The root capability could not be opened. + Io(std::io::Error), + /// The listener could not be bound. + Listener(String), +} + +impl std::fmt::Display for ServerBuildError { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Config(error) => write!(formatter, "invalid server configuration: {error}"), + Self::Io(error) => write!(formatter, "filesystem root could not be opened: {error}"), + Self::Listener(error) => write!(formatter, "HTTP listener could not be bound: {error}"), + } + } +} + +impl std::error::Error for ServerBuildError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self { + Self::Config(error) => Some(error), + Self::Io(error) => Some(error), + Self::Listener(_) => None, + } + } +} + +impl From for ServerBuildError { + fn from(value: ConfigError) -> Self { + Self::Config(value) + } +} + +impl From for ServerBuildError { + fn from(value: std::io::Error) -> Self { + Self::Io(value) + } +} diff --git a/crates/wip-fs-server/tests/end_to_end.rs b/crates/wip-fs-server/tests/end_to_end.rs new file mode 100644 index 0000000..9e44542 --- /dev/null +++ b/crates/wip-fs-server/tests/end_to_end.rs @@ -0,0 +1,333 @@ +use std::collections::BTreeMap; +use std::io::{Read, Write}; +use std::net::{SocketAddr, TcpStream}; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::thread; +use std::time::Duration; + +use tempfile::tempdir; +use wip_fs_server::{AccessMode, ServerConfig, TransportSecurity, WipFsServer}; +use wip_https::{ + BatchRequest, BatchRequestItem, BatchRequestKind, BatchResponse, BatchSuccess, DecodedResponse, + Endpoint, JSON_CONTENT_TYPE, Route, decode_batch_request, decode_batch_response, + decode_call_operation_response, decode_fetch_interface_response, decode_fetch_response, + decode_fetch_tree_response, encode_batch_request, encode_call_operation_request, + encode_fetch_interface_request, encode_fetch_request, encode_fetch_tree_request, +}; +use wip_protocol::{ + CallOperationRequest, FetchInterfaceRequest, FetchRequest, FetchTreeRequest, InterfaceTarget, + Target, Value, +}; + +#[test] +fn real_listener_serves_four_core_interactions_batch_and_write() { + let temporary = tempdir().expect("temporary root"); + std::fs::write(temporary.path().join("hello.txt"), b"hello world").expect("write file"); + std::fs::create_dir(temporary.path().join("nested")).expect("create directory"); + + let mut config = ServerConfig::for_root(temporary.path()); + config.access = AccessMode::ReadWrite; + config.transport_security = TransportSecurity::DevelopmentHttp; + let limits = config.limits.wire; + let bound = Arc::new( + WipFsServer::new(config) + .expect("build server") + .bind() + .expect("bind"), + ); + let address = bound.local_addr(); + let shutdown = Arc::new(AtomicBool::new(false)); + let worker = { + let bound = Arc::clone(&bound); + let shutdown = Arc::clone(&shutdown); + thread::spawn(move || bound.serve_until(shutdown)) + }; + let endpoint = Endpoint::parse(&format!("https://localhost:{}/wip", address.port())) + .expect("client endpoint"); + + let file_request = FetchRequest { + path: "/hello.txt".into(), + }; + let response = exchange( + address, + encode_fetch_request(&endpoint, &file_request, limits).expect("encode fetch"), + ); + let file = + success(decode_fetch_response(&file_request, &response, limits).expect("decode fetch")); + assert_eq!(file.name, "hello.txt"); + assert_eq!(file.interfaces.len(), 2); + assert!(file.validator.is_some()); + + let tree_request = FetchTreeRequest { + path: "/".into(), + depth: 1, + limit: 10, + }; + let response = exchange( + address, + encode_fetch_tree_request(&endpoint, &tree_request, limits).expect("encode tree"), + ); + let tree = + success(decode_fetch_tree_response(&tree_request, &response, limits).expect("decode tree")); + assert!(!tree.truncated); + assert_eq!(tree.root.children.len(), 2); + + let read_interface_request = FetchInterfaceRequest { + interface: file.interfaces[0].clone(), + }; + let response = exchange( + address, + encode_fetch_interface_request(&endpoint, &read_interface_request, limits) + .expect("encode interface fetch"), + ); + let read_interface = success( + decode_fetch_interface_response(&read_interface_request, &response, limits) + .expect("decode interface"), + ); + assert_eq!(read_interface.descriptor.operations[0].name, "read"); + + let read_request = CallOperationRequest { + target: Target { + path: "/hello.txt".into(), + validator: file.validator.clone(), + }, + interface: InterfaceTarget { + reference: file.interfaces[0].clone(), + validator: None, + }, + operation: "read".into(), + arguments: BTreeMap::from([ + ("offset".into(), Value::Integer(6)), + ("length".into(), Value::Integer(5)), + ]), + }; + let response = exchange( + address, + encode_call_operation_request(&endpoint, &read_request, &read_interface.descriptor, limits) + .expect("encode read call"), + ); + let read = success( + decode_call_operation_response( + &read_request, + &read_interface.descriptor, + &response, + limits, + ) + .expect("decode read call"), + ); + assert_eq!(read.result, Value::Bytes(b"world".to_vec())); + + let batch_request = BatchRequest { + requests: vec![ + BatchRequestItem { + id: "root".into(), + request: BatchRequestKind::Fetch(FetchRequest { path: "/".into() }), + }, + BatchRequestItem { + id: "tree".into(), + request: BatchRequestKind::FetchTree(FetchTreeRequest { + path: "/nested".into(), + depth: 1, + limit: 4, + }), + }, + BatchRequestItem { + id: "interface".into(), + request: BatchRequestKind::FetchInterface(read_interface_request.clone()), + }, + ], + }; + let response = exchange( + address, + encode_batch_request(&endpoint, &batch_request, limits).expect("encode batch"), + ); + let batch = + success(decode_batch_response(&batch_request, &response, limits).expect("decode batch")); + assert_batch_successes(&batch); + + let write_interface_request = FetchInterfaceRequest { + interface: file.interfaces[1].clone(), + }; + let response = exchange( + address, + encode_fetch_interface_request(&endpoint, &write_interface_request, limits) + .expect("encode write interface fetch"), + ); + let write_interface = success( + decode_fetch_interface_response(&write_interface_request, &response, limits) + .expect("decode write interface"), + ); + let write_request = CallOperationRequest { + target: Target { + path: "/hello.txt".into(), + validator: file.validator, + }, + interface: InterfaceTarget { + reference: file.interfaces[1].clone(), + validator: None, + }, + operation: "replace".into(), + arguments: BTreeMap::from([("contents".into(), Value::Bytes(b"updated".to_vec()))]), + }; + let response = exchange( + address, + encode_call_operation_request( + &endpoint, + &write_request, + &write_interface.descriptor, + limits, + ) + .expect("encode write call"), + ); + let write = success( + decode_call_operation_response( + &write_request, + &write_interface.descriptor, + &response, + limits, + ) + .expect("decode write call"), + ); + assert!(write.validator.is_some()); + assert_eq!( + std::fs::read(temporary.path().join("hello.txt")).expect("read updated file"), + b"updated" + ); + + shutdown.store(true, Ordering::Release); + worker + .join() + .expect("join listener") + .expect("listener exits cleanly"); +} + +#[test] +fn batch_keeps_known_malformed_arguments_item_local() { + let temporary = tempdir().expect("temporary root"); + let config = ServerConfig::for_root(temporary.path()); + let limits = config.limits.wire; + let server = WipFsServer::new(config).expect("build server"); + let endpoint = Endpoint::parse("https://localhost/wip").expect("endpoint"); + let body = br#"{"requests":[{"id":"ok","method":"fetch","arguments":{"path":"/"}},{"id":"bad","method":"fetch_tree","arguments":{"path":"/","depth":"bad","limit":1}}]}"#; + let logical = decode_batch_request(body, limits).expect("decode item-local invalid request"); + assert!(matches!( + logical.requests[1].request, + BatchRequestKind::Invalid { .. } + )); + let request = http::Request::builder() + .method(http::Method::POST) + .uri(endpoint.route(Route::Batch).as_str()) + .header(http::header::CONTENT_TYPE, JSON_CONTENT_TYPE) + .header(http::header::ACCEPT, "application/json") + .body(body.to_vec()) + .expect("build request"); + let response = server.handle_http(request); + let batch = success( + decode_batch_response(&logical, &response, limits).expect("decode mixed batch response"), + ); + assert!(matches!( + batch.responses[0].outcome, + Ok(BatchSuccess::Fetch(_)) + )); + assert!(batch.responses[1].outcome.is_err()); +} + +#[test] +fn call_argument_decoding_does_not_mask_missing_target() { + let temporary = tempdir().expect("temporary root"); + let config = ServerConfig::for_root(temporary.path()); + let server = WipFsServer::new(config).expect("build server"); + let endpoint = Endpoint::parse("https://localhost/wip").expect("endpoint"); + let body = br#"{"target":{"path":"/missing"},"interface":{"reference":"unknown"},"operation":"missing","arguments":{"bytes":"not-base64"}}"#; + let request = http::Request::builder() + .method(http::Method::POST) + .uri(endpoint.route(Route::CallOperation).as_str()) + .header(http::header::CONTENT_TYPE, JSON_CONTENT_TYPE) + .header(http::header::ACCEPT, "application/json") + .body(body.to_vec()) + .expect("build request"); + let response = server.handle_http(request); + assert_eq!(response.status(), http::StatusCode::NOT_FOUND); + assert!( + std::str::from_utf8(response.body()) + .expect("UTF-8 error") + .contains("\"not_found\"") + ); +} + +fn assert_batch_successes(batch: &BatchResponse) { + assert_eq!(batch.responses.len(), 3); + assert!(matches!( + batch.responses[0].outcome, + Ok(BatchSuccess::Fetch(_)) + )); + assert!(matches!( + batch.responses[1].outcome, + Ok(BatchSuccess::FetchTree(_)) + )); + assert!(matches!( + batch.responses[2].outcome, + Ok(BatchSuccess::FetchInterface(_)) + )); +} + +fn success(response: DecodedResponse) -> T { + match response { + DecodedResponse::Success(value) => value, + DecodedResponse::ProtocolFailure { error, .. } => panic!("protocol failure: {error}"), + } +} + +fn exchange(address: SocketAddr, request: http::Request>) -> http::Response> { + let mut stream = TcpStream::connect(address).expect("connect listener"); + stream + .set_read_timeout(Some(Duration::from_secs(5))) + .expect("set read timeout"); + let path = request + .uri() + .path_and_query() + .map_or("/", |value| value.as_str()); + write!( + stream, + "{} {} HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n", + request.method(), + path + ) + .expect("write request line"); + for (name, value) in request.headers() { + write!( + stream, + "{}: {}\r\n", + name, + value.to_str().expect("ASCII request header") + ) + .expect("write header"); + } + write!(stream, "Content-Length: {}\r\n\r\n", request.body().len()) + .expect("write content length"); + stream.write_all(request.body()).expect("write body"); + stream.flush().expect("flush request"); + + let mut raw = Vec::new(); + stream.read_to_end(&mut raw).expect("read response"); + let split = raw + .windows(4) + .position(|window| window == b"\r\n\r\n") + .expect("HTTP response header"); + let head = std::str::from_utf8(&raw[..split]).expect("UTF-8 response head"); + let mut lines = head.split("\r\n"); + let status = lines + .next() + .and_then(|line| line.split_whitespace().nth(1)) + .and_then(|status| status.parse::().ok()) + .expect("response status"); + let mut builder = http::Response::builder().status(status); + for line in lines { + let (name, value) = line.split_once(':').expect("response header"); + builder = builder.header(name, value.trim()); + } + builder + .body(raw[split + 4..].to_vec()) + .expect("build response") +}