From b196047b5765b6f930c531134859936e7b483f1d Mon Sep 17 00:00:00 2001 From: Hare Date: Tue, 6 Oct 2026 18:12:05 +0900 Subject: [PATCH] feat: expose extensible filesystem projection engine Separate pinned Linux backend, typed composition, coherent principal policy views and standard HTTP adapter. Refresh validators inside serialized execution and preserve default listener and effect boundaries. Linux/adapter and real-listener regressions pass. --- crates/wip-fs-server/Cargo.toml | 2 +- crates/wip-fs-server/src/defaults.rs | 146 ++++ crates/wip-fs-server/src/engine.rs | 594 ++++++++++++++++ crates/wip-fs-server/src/http_adapter.rs | 435 ++++++++++++ crates/wip-fs-server/src/lib.rs | 46 +- crates/wip-fs-server/src/projection.rs | 825 ++++++++++++++--------- crates/wip-fs-server/src/server.rs | 222 +----- 7 files changed, 1731 insertions(+), 539 deletions(-) create mode 100644 crates/wip-fs-server/src/defaults.rs create mode 100644 crates/wip-fs-server/src/engine.rs create mode 100644 crates/wip-fs-server/src/http_adapter.rs diff --git a/crates/wip-fs-server/Cargo.toml b/crates/wip-fs-server/Cargo.toml index f230e0f..882f70c 100644 --- a/crates/wip-fs-server/Cargo.toml +++ b/crates/wip-fs-server/Cargo.toml @@ -3,7 +3,7 @@ name = "wip-fs-server" version.workspace = true edition.workspace = true rust-version.workspace = true -description = "Request-time filesystem projection server for WIP over HTTP" +description = "Extensible request-time filesystem projection library and WIP over HTTP server" license.workspace = true repository.workspace = true readme = "README.md" diff --git a/crates/wip-fs-server/src/defaults.rs b/crates/wip-fs-server/src/defaults.rs new file mode 100644 index 0000000..bb39192 --- /dev/null +++ b/crates/wip-fs-server/src/defaults.rs @@ -0,0 +1,146 @@ +//! Reusable default Linux composition; no listener or authentication dependency. +use crate::engine::{BackendEntry, Composition, DefaultPolicy, EntryKind, Projection}; +use crate::interfaces::{ + FileRead, FileReadDefinition, FileWrite, FileWriteDefinition, SymlinkInspect, + SymlinkInspectDefinition, +}; +use crate::projection::{LinuxBackend, LinuxEntry}; +use crate::{AccessMode, ResourceLimits}; +use wip_host::{ + ExecutionContext, HostError, InterfaceRef, InterfaceRegistry, PublishedObject, + RegisteredInterface, RegistrationError, Validator, WipObject, +}; +use wip_interface as wip; + +/// Registered built-in interfaces, usable by external compositions and Objects. +pub struct DefaultInterfaces { + /// Bounded regular-file reads. + pub read: RegisteredInterface, + /// No-follow link inspection. + pub symlink: RegisteredInterface, + /// Validator-protected replacement. + pub write: RegisteredInterface, +} +impl DefaultInterfaces { + /// Register reusable default descriptors without publishing any objects. + pub fn register(registry: &mut InterfaceRegistry) -> Result { + Ok(Self { + read: registry.register(FileReadDefinition, None)?, + symlink: registry.register(SymlinkInspectDefinition, None)?, + write: registry.register(FileWriteDefinition, None)?, + }) + } +} + +/// Default capability-aware Linux publication, independent of HTTP. +pub struct DefaultComposition { + interfaces: DefaultInterfaces, +} +impl DefaultComposition { + /// Use previously registered built-in interface tokens. + pub fn new(interfaces: DefaultInterfaces) -> Self { + Self { interfaces } + } +} + +/// Request-scoped default Object; external Objects may instead wrap `LinuxEntry`. +pub struct DefaultObject { + entry: LinuxEntry, + write: InterfaceRef, +} +impl DefaultObject { + /// The pinned, bounded capability used by every operation on this Object. + pub fn entry(&self) -> &LinuxEntry { + &self.entry + } +} +impl WipObject for DefaultObject { + fn name(&self) -> &str { + self.entry.name() + } + fn description(&self) -> Option<&str> { + self.entry.description() + } + fn reference(&self) -> Option<&str> { + self.entry.reference() + } + fn validator(&self) -> Option<&Validator> { + self.entry.validator() + } + fn requires_validator(&self, interface: &InterfaceRef, operation: &str) -> bool { + interface == &self.write && operation == "replace" + } +} +impl FileRead for DefaultObject { + fn read(&self, offset: Option, length: Option) -> wip::OperationResult { + self.entry.read(offset, length) + } +} +impl SymlinkInspect for DefaultObject { + fn read_link(&self) -> wip::OperationResult { + self.entry.read_link() + } +} +impl FileWrite for DefaultObject { + fn replace(&self, contents: wip::Bytes) -> wip::OperationResult<()> { + self.entry.replace(contents) + } +} +impl Composition for DefaultComposition { + type Object = DefaultObject; + fn publish( + &self, + entry: LinuxEntry, + _: &ExecutionContext, + ) -> Result, HostError> { + let kind = entry.kind(); + let writable = entry.access_mode() == AccessMode::ReadWrite; + let mut publication = PublishedObject::new(DefaultObject { + entry, + write: self.interfaces.write.reference().clone(), + }); + match kind { + EntryKind::Regular => { + publication = publication + .expose(&self.interfaces.read) + .map_err(publication_error)?; + if writable { + publication = publication + .expose(&self.interfaces.write) + .map_err(publication_error)?; + } + } + EntryKind::Symlink => { + publication = publication + .expose(&self.interfaces.symlink) + .map_err(publication_error)?; + } + EntryKind::Directory | EntryKind::Unsupported => {} + } + Ok(publication) + } +} +fn publication_error(_: wip_host::PublicationError) -> HostError { + HostError::internal( + "default publication failed", + wip_host::EffectGuarantee::NotCommitted, + ) +} + +/// Construct the secure-by-default Linux engine without binding a listener. +pub fn default_projection( + root: &std::path::Path, + access: AccessMode, + limits: ResourceLimits, +) -> Result, std::io::Error> { + let backend = LinuxBackend::open(root, access, limits)?; + let mut registry = InterfaceRegistry::new(); + let interfaces = DefaultInterfaces::register(&mut registry).map_err(std::io::Error::other)?; + Ok(Projection::new( + backend, + registry, + DefaultComposition::new(interfaces), + DefaultPolicy, + limits, + )) +} diff --git a/crates/wip-fs-server/src/engine.rs b/crates/wip-fs-server/src/engine.rs new file mode 100644 index 0000000..e00224d --- /dev/null +++ b/crates/wip-fs-server/src/engine.rs @@ -0,0 +1,594 @@ +//! Transport-independent projection, publication, and principal view boundaries. + +use crate::ResourceLimits; +use wip_host::{ + EffectGuarantee, ExecutionContext, HostError, InterfaceRef, InterfaceRegistry, ObjectProvider, + OperationAttempt, ProviderResult, PublishedObject, WipObject, +}; +use wip_protocol::{ + FetchInterfaceRequest, FetchInterfaceResponse, ObjectObservation, ObserveRequest, + ObserveResponse, ProtocolError, ProtocolErrorCode, +}; + +/// Portable entry classification; capabilities must not be inferred from kind alone. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum EntryKind { + /// Container whose children can be enumerated. + Directory, + /// Regular data entry. + Regular, + /// Link itself, never implicitly followed. + Symlink, + /// Metadata-only entry unless the composition supplies a capability. + Unsupported, +} + +/// A root-confined, request-time capability, not a path to resolve again. +/// +/// Metadata must be safe for public exposure. Validators bind both identity and +/// protected state. Cloning must retain the same target rather than re-resolve it. +pub trait BackendEntry: WipObject + Clone { + /// Classification for observation and publication selection. + fn kind(&self) -> EntryKind; +} + +/// Trusted backend boundary. Implementations need not use POSIX semantics. +/// +/// Resolve and enumerate pinned capabilities within the configured root. Enforce +/// context cancellation, deadlines and work budgets during backend work. Do not +/// advertise capabilities the backend cannot guarantee. Errors must not include +/// backend paths, credentials or hidden metadata. +/// +/// `execute` must retain the same target and serialize validator observation, +/// publication, validation, typed dispatch and commit/rollback. Refresh entry +/// validators *inside* the boundary before invoking the callback, and return the +/// committed post-operation validator on success. Never upgrade an unknown +/// outcome except after confirmed rollback. Locks only protect against actors +/// participating in that locking protocol; document any external-writer limits. +pub trait FilesystemBackend { + /// Stable capability supplied to the composition. + type Entry: BackendEntry; + /// Resolve a canonical WIP path without escaping the configured root. + fn resolve(&self, context: &ExecutionContext, path: &str) -> ProviderResult; + /// Enumerate direct children, at most `maximum`; excess must fail, not truncate. + fn children( + &self, + context: &ExecutionContext, + entry: &Self::Entry, + maximum: usize, + ) -> Result, HostError>; + /// Execute against this already-resolved target within an atomic boundary. + fn execute( + &self, + context: &ExecutionContext, + entry: Self::Entry, + operation: impl FnOnce(Self::Entry) -> OperationAttempt, + ) -> ProviderResult; +} + +/// Typed publication extension point, implementable on an external crate's type. +/// +/// Use `InterfaceRegistry::register` and `PublishedObject::expose` with generated +/// definitions. An external local Object can wrap the entry and implement any +/// number of interfaces, avoiding orphan rules. Its `WipObject` implementation +/// declares validator requirements per qualified operation. Operation failures +/// declare their own `EffectGuarantee`, not a fixed read/write classification. +pub trait Composition { + /// Local typed operation implementation and safe public metadata. + type Object: WipObject; + /// Build a request-scoped publication using this exact capability/context. + /// + /// Publication and policy hooks must be side-effect-free with respect to the + /// backend. A publication failure is a pre-dispatch rejection, never a + /// started Operation. Delegate metadata/validators to the boundary snapshot. + fn publish( + &self, + entry: E, + context: &ExecutionContext, + ) -> Result, HostError>; +} + +/// Principal-specific view policy, independent of authentication transport. +/// +/// Keep decisions deterministic and stable within an effective Worldspace view. +/// A changed policy requires a new view/reference lifetime (or interface +/// validators). Operation visibility is interface-wide within a view so every +/// published reference has exactly the descriptor returned by fetch. Entry- +/// specific differences should select different registered interfaces. Hooks +/// cannot add implementations, metadata, or raw paths to a publication. +pub trait ProjectionPolicy { + /// Opaque deployment-established principal/request identity. + type Principal; + /// Whether the entry is visible. Invisible ancestors hide their whole subtree. + fn object_visible(&self, principal: &Self::Principal, entry: &E) -> bool; + /// Whether an interface descriptor may be fetched in this view. + fn interface_visible(&self, _principal: &Self::Principal, _interface: &InterfaceRef) -> bool { + true + } + /// Select/order a subset of implemented interfaces; defaults to all. + fn interfaces( + &self, + _principal: &Self::Principal, + _entry: &E, + available: &[InterfaceRef], + ) -> Vec { + available.to_vec() + } + /// Whether an operation is present in this view's descriptor and dispatch. + fn operation_visible( + &self, + _principal: &Self::Principal, + _interface: &InterfaceRef, + _operation: &str, + ) -> bool { + true + } + /// Execution permission for a visible operation, after canonical preconditions. + fn authorize( + &self, + _principal: &Self::Principal, + _object: &O, + _interface: &InterfaceRef, + _operation: &str, + ) -> bool { + true + } +} + +/// No authentication assumption: all configured objects/operations are visible. +pub struct DefaultPolicy; +impl ProjectionPolicy for DefaultPolicy { + type Principal = (); + fn object_visible(&self, _: &(), _: &E) -> bool { + true + } +} + +/// Canonical request-time engine, independent of HTTP and Linux. +pub struct Projection { + backend: B, + registry: InterfaceRegistry, + composition: C, + policy: P, + limits: ResourceLimits, +} + +impl Projection +where + B: FilesystemBackend, + C: Composition, + P: ProjectionPolicy, +{ + /// Assemble trusted components. The registry must contain all exposed tokens. + pub fn new( + backend: B, + registry: InterfaceRegistry, + composition: C, + policy: P, + limits: ResourceLimits, + ) -> Self { + Self { + backend, + registry, + composition, + policy, + limits, + } + } + /// Backend access for trusted composition/configuration, never serialized. + pub fn backend(&self) -> &B { + &self.backend + } + /// Borrow one effective principal view for observation, fetch and dispatch. + pub fn view<'a>(&'a self, principal: &'a P::Principal) -> ProjectionView<'a, B, C, P> { + ProjectionView { + projection: self, + principal, + } + } +} + +/// A coherent principal view implementing canonical typed host dispatch. +pub struct ProjectionView<'a, B, C, P> +where + B: FilesystemBackend, + C: Composition, + P: ProjectionPolicy, +{ + projection: &'a Projection, + principal: &'a P::Principal, +} + +impl ProjectionView<'_, B, C, P> +where + B: FilesystemBackend, + C: Composition, + P: ProjectionPolicy, +{ + fn resolve(&self, context: &ExecutionContext, path: &str) -> ProviderResult { + if let Err(error) = preflight(context) { + return ProviderResult::Failure(error); + } + if wip_protocol::validate_path(path).is_err() { + return ProviderResult::NotFound; + } + // Check ancestors as well: guessing a descendant cannot bypass hidden parents. + let mut end = 1; + loop { + let prefix = if end == 1 { "/" } else { &path[..end] }; + let entry = match self.projection.backend.resolve(context, prefix) { + ProviderResult::Found(entry) => entry, + ProviderResult::NotFound => return ProviderResult::NotFound, + ProviderResult::Failure(error) => return ProviderResult::Failure(error), + }; + if !self + .projection + .policy + .object_visible(self.principal, &entry) + { + return ProviderResult::NotFound; + } + if prefix == path { + return ProviderResult::Found(entry); + } + let start = if end == 1 { 1 } else { end + 1 }; + end = path[start..] + .find('/') + .map_or(path.len(), |index| start + index); + } + } + + fn publication( + &self, + entry: B::Entry, + context: &ExecutionContext, + ) -> Result, HostError> { + let publication = self + .projection + .composition + .publish(entry.clone(), context)?; + let available = publication + .protocol_object() + .interfaces + .iter() + .map(|reference| { + self.projection + .registry + .resolve(reference) + .ok_or_else(internal) + }) + .collect::, _>>()?; + let mut selected = self + .projection + .policy + .interfaces(self.principal, &entry, &available); + selected.retain(|reference| { + self.projection + .policy + .interface_visible(self.principal, reference) + }); + publication + .project(&selected, |reference, operation| { + self.projection + .policy + .operation_visible(self.principal, reference, operation) + }) + .map_err(|_| internal()) + } + + /// Retrieve the descriptor with precisely this principal's visible operations. + pub fn fetch_interface( + &self, + request: &FetchInterfaceRequest, + ) -> Result { + read_boundary(|| self.fetch_interface_inner(request)) + } + + fn fetch_interface_inner( + &self, + request: &FetchInterfaceRequest, + ) -> Result { + let observation = self + .projection + .registry + .fetch_interface(&request.interface)?; + if !self + .projection + .policy + .interface_visible(self.principal, &observation.reference) + { + return Err(protocol( + ProtocolErrorCode::InterfaceNotFound, + "interface is not published", + )); + } + let reference = observation.reference.clone(); + observation + .project_operations(|operation| { + self.projection + .policy + .operation_visible(self.principal, &reference, operation) + }) + .protocol_response() + } + + /// Observe a complete bounded tree, omitting invisible children and subtrees. + pub fn observe( + &self, + context: &ExecutionContext, + request: &ObserveRequest, + ) -> Result { + read_boundary(|| self.observe_inner(context, request)) + } + + fn observe_inner( + &self, + context: &ExecutionContext, + request: &ObserveRequest, + ) -> Result { + request + .validate() + .map_err(|_| protocol(ProtocolErrorCode::InvalidRequest, "invalid observe request"))?; + if request.depth > self.projection.limits.max_observation_depth { + return Err(resource_protocol()); + } + let entry = match self.resolve(context, &request.path) { + ProviderResult::Found(entry) => entry, + ProviderResult::NotFound => { + return Err(protocol( + ProtocolErrorCode::NotFound, + "target path is not published", + )); + } + ProviderResult::Failure(error) => return Err(read_error(error)), + }; + let mut remaining = u64::from(self.projection.limits.max_observation_nodes) + .min(context.resource_budget().unwrap_or(u64::MAX)); + let response = self.observe_entry(context, entry, request.depth, &mut remaining)?; + request.validate_response(&response).map_err(|_| { + protocol( + ProtocolErrorCode::Internal, + "invalid projection observation", + ) + })?; + Ok(response) + } + + fn observe_entry( + &self, + context: &ExecutionContext, + entry: B::Entry, + depth: u32, + remaining: &mut u64, + ) -> Result { + preflight(context).map_err(read_error)?; + if *remaining == 0 { + return Err(resource_protocol()); + } + *remaining -= 1; + let object = self + .publication(entry.clone(), context) + .map_err(read_error)? + .protocol_object(); + if depth == 0 { + return Ok(ObjectObservation { + object, + children: None, + }); + } + let mut children = Vec::new(); + if entry.kind() == EntryKind::Directory { + let entries = self + .projection + .backend + .children( + context, + &entry, + usize::try_from(*remaining).unwrap_or(usize::MAX), + ) + .map_err(read_error)?; + if entries.len() as u64 > *remaining { + return Err(resource_protocol()); + } + for child in entries { + if self + .projection + .policy + .object_visible(self.principal, &child) + { + children.push(self.observe_entry(context, child, depth - 1, remaining)?); + } else { + if *remaining == 0 { + return Err(resource_protocol()); + } + *remaining -= 1; + } + } + } + Ok(ObjectObservation { + object, + children: Some(children), + }) + } +} + +impl ObjectProvider for ProjectionView<'_, B, C, P> +where + B: FilesystemBackend, + C: Composition, + P: ProjectionPolicy, +{ + type Object = C::Object; + fn with_object( + &self, + context: &ExecutionContext, + path: &str, + operation: impl FnOnce(&PublishedObject) -> R, + ) -> ProviderResult { + match self.resolve(context, path) { + ProviderResult::Found(entry) => match self.publication(entry, context) { + Ok(publication) => ProviderResult::Found(operation(&publication)), + Err(error) => ProviderResult::Failure(error), + }, + ProviderResult::NotFound => ProviderResult::NotFound, + ProviderResult::Failure(error) => ProviderResult::Failure(error), + } + } + fn execute_operation( + &self, + context: &ExecutionContext, + path: &str, + operation: impl FnOnce(&PublishedObject) -> OperationAttempt, + ) -> ProviderResult { + let entry = match self.resolve(context, path) { + ProviderResult::Found(entry) => entry, + ProviderResult::NotFound => return ProviderResult::NotFound, + ProviderResult::Failure(error) => return ProviderResult::Failure(error), + }; + self.projection.backend.execute(context, entry, |entry| { + // Recheck visibility using the atomic-boundary snapshot. + // Failure here is still pre-dispatch, with no operation effects. + if !self + .projection + .policy + .object_visible(self.principal, &entry) + { + return OperationAttempt::reject(wip_host::CallErrorKind::NotFound); + } + if let Err(error) = preflight(context) { + return OperationAttempt::reject(match error.kind() { + wip_host::HostErrorKind::Internal => wip_host::CallErrorKind::Internal, + wip_host::HostErrorKind::ResourceLimitExceeded => { + wip_host::CallErrorKind::ResourceLimitExceeded + } + }); + } + match self.publication(entry, context) { + Ok(publication) => operation(&publication), + Err(error) => OperationAttempt::reject(match error.kind() { + wip_host::HostErrorKind::Internal => wip_host::CallErrorKind::Internal, + wip_host::HostErrorKind::ResourceLimitExceeded => { + wip_host::CallErrorKind::ResourceLimitExceeded + } + }), + } + }) + } + fn authorize( + &self, + _: &str, + object: &Self::Object, + interface: &InterfaceRef, + operation: &str, + ) -> bool { + self.projection + .policy + .authorize(self.principal, object, interface, operation) + } +} + +impl ObjectProvider for Projection +where + B: FilesystemBackend, + C: Composition, + P: ProjectionPolicy, +{ + type Object = C::Object; + fn with_object( + &self, + context: &ExecutionContext, + path: &str, + operation: impl FnOnce(&PublishedObject) -> R, + ) -> ProviderResult { + self.view(&()).with_object(context, path, operation) + } + fn execute_operation( + &self, + context: &ExecutionContext, + path: &str, + operation: impl FnOnce(&PublishedObject) -> OperationAttempt, + ) -> ProviderResult { + self.view(&()).execute_operation(context, path, operation) + } + fn authorize( + &self, + path: &str, + object: &Self::Object, + interface: &InterfaceRef, + operation: &str, + ) -> bool { + self.view(&()).authorize(path, object, interface, operation) + } +} + +impl Projection +where + B: FilesystemBackend, + C: Composition, + P: ProjectionPolicy, +{ + /// Observe the default unit-principal view. + pub fn observe( + &self, + context: &ExecutionContext, + request: &ObserveRequest, + ) -> Result { + self.view(&()).observe(context, request) + } + /// Fetch in the default unit-principal view. + pub fn fetch_interface( + &self, + request: &FetchInterfaceRequest, + ) -> Result { + self.view(&()).fetch_interface(request) + } +} + +fn read_boundary( + operation: impl FnOnce() -> Result, +) -> Result { + std::panic::catch_unwind(std::panic::AssertUnwindSafe(operation)) + .unwrap_or_else(|_| Err(protocol(ProtocolErrorCode::Internal, "projection failed"))) +} + +fn internal() -> HostError { + HostError::internal( + "projection publication failed", + EffectGuarantee::NotCommitted, + ) +} +fn protocol(code: ProtocolErrorCode, message: &str) -> ProtocolError { + ProtocolError { + code, + message: message.into(), + } +} +fn resource_protocol() -> ProtocolError { + protocol( + ProtocolErrorCode::ResourceLimitExceeded, + "projection resource limit exceeded", + ) +} +fn read_error(error: HostError) -> ProtocolError { + match error.kind() { + wip_host::HostErrorKind::ResourceLimitExceeded => resource_protocol(), + wip_host::HostErrorKind::Internal => { + protocol(ProtocolErrorCode::Internal, "projection failed") + } + } +} +fn preflight(context: &ExecutionContext) -> Result<(), HostError> { + if context.cancellation().is_cancelled() + || context + .deadline() + .is_some_and(|deadline| std::time::Instant::now() >= deadline) + || context.resource_budget() == Some(0) + { + Err(HostError::resource_limit( + "projection work limit exceeded", + EffectGuarantee::NotCommitted, + )) + } else { + Ok(()) + } +} diff --git a/crates/wip-fs-server/src/http_adapter.rs b/crates/wip-fs-server/src/http_adapter.rs new file mode 100644 index 0000000..bf16db7 --- /dev/null +++ b/crates/wip-fs-server/src/http_adapter.rs @@ -0,0 +1,435 @@ +//! Framework-neutral standard WIP over HTTP mapping for a projection view. + +use std::time::Instant; + +use http::header::ALLOW; +use http::{Request, Response, StatusCode}; +use wip_host::{ExecutionContext, ObjectProvider}; +use wip_http::{ + CodecError, EncodeHttpError, Endpoint, RequestMetadataError, Route, + decode_call_operation_metadata, decode_fetch_interface_request, decode_observe_request, + encode_call_operation_response, encode_fetch_interface_response, encode_observe_response, + encode_protocol_error_response, validate_http_request, +}; +use wip_protocol::{ProtocolError, ProtocolErrorCode, ProtocolInteraction}; + +use crate::engine::ProjectionView; +use crate::{Composition, FilesystemBackend, ProjectionPolicy, ResourceLimits}; + +/// Standard HTTP routes and codecs over a caller-scoped projection view. +/// +/// The adapter borrows the view and endpoint without owning a listener, selecting +/// a principal, or imposing a new execution context. Integrations must enforce +/// the request body limit while collecting bytes, before calling this adapter. +pub struct HttpAdapter<'a, 'view, B, C, P> +where + B: FilesystemBackend, + C: Composition, + P: ProjectionPolicy, +{ + view: &'a ProjectionView<'view, B, C, P>, + endpoint: &'a Endpoint, + limits: ResourceLimits, +} + +impl<'a, 'view, B, C, P> HttpAdapter<'a, 'view, B, C, P> +where + B: FilesystemBackend, + C: Composition, + P: ProjectionPolicy, +{ + /// Creates an adapter for an already-selected principal's projection view. + #[must_use] + pub fn new( + view: &'a ProjectionView<'view, B, C, P>, + endpoint: &'a Endpoint, + limits: ResourceLimits, + ) -> Self { + Self { + view, + endpoint, + limits, + } + } + + /// Handles one buffered request using the supplied execution controls. + /// + /// Observation and operation dispatch receive the original context, including + /// cancellation and resource budget. An elapsed deadline after dispatch keeps + /// the listener's existing empty HTTP 504 transport-failure behavior. + #[must_use] + pub fn handle_http_with_context( + &self, + request: Request>, + context: &ExecutionContext, + ) -> Response> { + let Some(route) = self.route(request.uri()) else { + return empty_response(StatusCode::NOT_FOUND); + }; + if let Err(error) = validate_http_request(self.endpoint, route, &request) { + return self.metadata_error(error); + } + let response = match route { + Route::Observe => self.observe(request.body(), context), + Route::FetchInterface => self.fetch_interface(request.body()), + Route::CallOperation => self.call_operation(request.body(), context), + }; + if context + .deadline() + .is_some_and(|deadline| Instant::now() >= deadline) + { + empty_response(StatusCode::GATEWAY_TIMEOUT) + } else { + response + } + } + + fn route(&self, uri: &http::Uri) -> Option { + if uri.query().is_some() { + return None; + } + [Route::Observe, Route::FetchInterface, Route::CallOperation] + .into_iter() + .find(|route| self.endpoint.route(*route).path() == uri.path()) + } + + fn observe(&self, body: &[u8], context: &ExecutionContext) -> Response> { + let request = match decode_observe_request(body, self.limits.wire) { + Ok(request) => request, + Err(error) => return self.codec_error(ProtocolInteraction::Observe, error), + }; + match self.view.observe(context, &request) { + Ok(tree) => { + encode_observe_response(&request, &tree, self.limits.wire).unwrap_or_else(|error| { + self.read_encode_error(ProtocolInteraction::Observe, error) + }) + } + Err(error) => self.protocol_error(ProtocolInteraction::Observe, &error), + } + } + + fn fetch_interface(&self, body: &[u8]) -> Response> { + let request = match decode_fetch_interface_request(body, self.limits.wire) { + Ok(request) => request, + Err(error) => return self.codec_error(ProtocolInteraction::FetchInterface, error), + }; + match self.view.fetch_interface(&request) { + Ok(interface) => { + encode_fetch_interface_response(&request, &interface, self.limits.wire) + .unwrap_or_else(|error| { + self.read_encode_error(ProtocolInteraction::FetchInterface, error) + }) + } + Err(error) => self.protocol_error(ProtocolInteraction::FetchInterface, &error), + } + } + + fn call_operation(&self, body: &[u8], context: &ExecutionContext) -> Response> { + let metadata = match decode_call_operation_metadata(body, self.limits.wire) { + Ok(metadata) => metadata, + Err(error) => return self.codec_error(ProtocolInteraction::CallOperation, error), + }; + let decoded = std::sync::Mutex::new(None); + let execution = self.view.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.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, + &operation_outcome_unknown(), + ); + }; + encode_call_operation_response(&request, &descriptor, &result, self.limits.wire) + .unwrap_or_else(|error| self.operation_encode_error(error)) + } + Err(error) => self.protocol_error(ProtocolInteraction::CallOperation, &error), + } + } + + fn operation_encode_error(&self, error: EncodeHttpError) -> Response> { + if matches!(&error, EncodeHttpError::Codec(codec) if is_transport_limit(codec)) { + return empty_response(StatusCode::INTERNAL_SERVER_ERROR); + } + // Any successfully executed operation may have committed effects. Its + // name cannot establish a non-commit guarantee after response failure. + self.protocol_error( + ProtocolInteraction::CallOperation, + &operation_outcome_unknown(), + ) + } + + fn metadata_error(&self, error: RequestMetadataError) -> Response> { + match error { + RequestMetadataError::NotFound => empty_response(StatusCode::NOT_FOUND), + error @ RequestMetadataError::MethodNotAllowed => Response::builder() + .status(error.status()) + .header( + ALLOW, + error + .allow() + .expect("method-not-allowed metadata has an Allow value"), + ) + .body(Vec::new()) + .expect("static method response is valid"), + 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 is_transport_limit(&error) { + empty_response(StatusCode::PAYLOAD_TOO_LARGE) + } else { + self.protocol_error( + interaction, + &ProtocolError { + code: ProtocolErrorCode::InvalidRequest, + message: "malformed WIP request".into(), + }, + ) + } + } + + fn read_encode_error( + &self, + interaction: ProtocolInteraction, + error: EncodeHttpError, + ) -> Response> { + match error { + EncodeHttpError::Codec(codec) if is_transport_limit(&codec) => { + empty_response(StatusCode::INTERNAL_SERVER_ERROR) + } + _ => self.protocol_error( + interaction, + &ProtocolError { + code: ProtocolErrorCode::Internal, + message: "HTTP adapter failed".into(), + }, + ), + } + } + + fn protocol_error( + &self, + interaction: ProtocolInteraction, + error: &ProtocolError, + ) -> Response> { + // Engine and policy diagnostics may contain paths, argument values, or + // backend details. Emit only fixed, bounded diagnostics on the wire. + let sanitized = ProtocolError { + code: error.code, + message: sanitized_message(error.code).into(), + }; + encode_protocol_error_response(interaction, &sanitized, self.limits.wire) + .unwrap_or_else(|_| empty_response(StatusCode::INTERNAL_SERVER_ERROR)) + } +} + +fn operation_outcome_unknown() -> ProtocolError { + ProtocolError { + code: ProtocolErrorCode::OperationOutcomeUnknown, + message: "operation completed but its response could not be encoded".into(), + } +} + +fn sanitized_message(code: ProtocolErrorCode) -> &'static str { + match code { + ProtocolErrorCode::InvalidRequest => "malformed WIP request", + ProtocolErrorCode::NotFound => "target path is not published", + ProtocolErrorCode::InterfaceNotFound => "interface is not published", + ProtocolErrorCode::InterfaceMismatch => "interface is not published on the target", + ProtocolErrorCode::OperationNotFound => "operation is not published", + ProtocolErrorCode::InvalidArguments => "arguments do not match the operation", + ProtocolErrorCode::ValidatorRequired => "object validator is required", + ProtocolErrorCode::ValidatorMismatch => "object validator does not match", + ProtocolErrorCode::InterfaceValidatorRequired => "interface validator is required", + ProtocolErrorCode::InterfaceValidatorMismatch => "interface validator does not match", + ProtocolErrorCode::PermissionDenied => "operation is not permitted", + ProtocolErrorCode::ResourceLimitExceeded => "projection resource limit exceeded", + ProtocolErrorCode::Internal => "HTTP adapter failed", + ProtocolErrorCode::OperationOutcomeUnknown => "operation outcome is unknown", + } +} + +fn is_transport_limit(error: &CodecError) -> bool { + matches!( + error, + CodecError::BodyTooLarge { .. } | CodecError::NestingTooDeep { .. } + ) +} + +fn empty_response(status: StatusCode) -> Response> { + Response::builder() + .status(status) + .body(Vec::new()) + .expect("static status builds an HTTP response") +} + +#[cfg(test)] +mod tests { + use super::*; + use tempfile::tempdir; + use wip_http::{ + DecodedResponse, Limits, decode_fetch_interface_response, decode_observe_response, + encode_fetch_interface_request, encode_observe_request, + }; + use wip_protocol::{FetchInterfaceRequest, ObserveRequest}; + + #[test] + fn adapter_serves_observation_and_interface_fetch_without_server() { + let root = tempdir().unwrap(); + std::fs::write(root.path().join("file"), b"contents").unwrap(); + let limits = ResourceLimits::default(); + let projection = + crate::default_projection(root.path(), crate::AccessMode::ReadOnly, limits).unwrap(); + let view = projection.view(&()); + let endpoint = Endpoint::parse("http://localhost/custom").unwrap(); + let adapter = HttpAdapter::new(&view, &endpoint, limits); + let context = ExecutionContext::new(); + let request = ObserveRequest { + path: "/file".into(), + depth: 0, + }; + let response = adapter.handle_http_with_context( + encode_observe_request(&endpoint, &request, limits.wire).unwrap(), + &context, + ); + let DecodedResponse::Success(observation) = + decode_observe_response(&request, &response, limits.wire).unwrap() + else { + panic!("expected observation success") + }; + let request = FetchInterfaceRequest { + interface: observation.object.interfaces[0].clone(), + }; + let response = adapter.handle_http_with_context( + encode_fetch_interface_request(&endpoint, &request, limits.wire).unwrap(), + &context, + ); + assert!(matches!( + decode_fetch_interface_response(&request, &response, limits.wire).unwrap(), + DecodedResponse::Success(_) + )); + } + + #[test] + fn adapter_preserves_transport_boundaries_and_execution_context() { + let root = tempdir().unwrap(); + let limits = ResourceLimits::default(); + let projection = + crate::default_projection(root.path(), crate::AccessMode::ReadOnly, limits).unwrap(); + let view = projection.view(&()); + let endpoint = Endpoint::parse("http://localhost/wip").unwrap(); + let adapter = HttpAdapter::new(&view, &endpoint, limits); + let context = ExecutionContext::new(); + for uri in [ + "/wip/v1/observe?query=1", + "/wip/v1/unknown", + "/wip/v1/observe/extra", + ] { + let response = adapter.handle_http_with_context( + Request::builder().uri(uri).body(Vec::new()).unwrap(), + &context, + ); + assert_eq!(response.status(), StatusCode::NOT_FOUND); + assert!(response.body().is_empty()); + } + let response = adapter.handle_http_with_context( + Request::builder() + .uri("/wip/v1/observe") + .body(Vec::new()) + .unwrap(), + &context, + ); + assert_eq!(response.status(), StatusCode::METHOD_NOT_ALLOWED); + assert_eq!(response.headers()[ALLOW], "POST"); + assert!(response.body().is_empty()); + let request = ObserveRequest { + path: "/".into(), + depth: 0, + }; + let response = adapter.handle_http_with_context( + encode_observe_request(&endpoint, &request, limits.wire).unwrap(), + &ExecutionContext::new().with_resource_budget(0), + ); + let text = std::str::from_utf8(response.body()).unwrap(); + assert!(text.contains("resource_limit_exceeded")); + let response = adapter.handle_http_with_context( + encode_observe_request(&endpoint, &request, limits.wire).unwrap(), + &ExecutionContext::new().with_deadline(Instant::now()), + ); + assert_eq!(response.status(), StatusCode::GATEWAY_TIMEOUT); + assert!(response.body().is_empty()); + } + + #[test] + fn post_success_encoding_is_unknown_except_empty_transport_limit_failure() { + let root = tempdir().unwrap(); + let limits = ResourceLimits::default(); + let projection = + crate::default_projection(root.path(), crate::AccessMode::ReadOnly, limits).unwrap(); + let view = projection.view(&()); + let endpoint = Endpoint::parse("http://localhost/wip").unwrap(); + let adapter = HttpAdapter::new(&view, &endpoint, limits); + let error = decode_observe_request(b"{", limits.wire).unwrap_err(); + let response = adapter.operation_encode_error(EncodeHttpError::Codec(error)); + assert!( + std::str::from_utf8(response.body()) + .unwrap() + .contains("operation_outcome_unknown") + ); + let tiny_limits = Limits::new(1, 1, 1).unwrap(); + let error = decode_observe_request(b"{}", tiny_limits).unwrap_err(); + assert!(is_transport_limit(&error)); + let response = adapter.operation_encode_error(EncodeHttpError::Codec(error)); + assert_eq!(response.status(), StatusCode::INTERNAL_SERVER_ERROR); + assert!(response.body().is_empty()); + assert!(!response.headers().contains_key(http::header::CONTENT_TYPE)); + } + + #[test] + fn engine_error_diagnostics_are_sanitized_and_bounded() { + let root = tempdir().unwrap(); + let limits = ResourceLimits::default(); + let projection = + crate::default_projection(root.path(), crate::AccessMode::ReadOnly, limits).unwrap(); + let view = projection.view(&()); + let endpoint = Endpoint::parse("http://localhost/wip").unwrap(); + let adapter = HttpAdapter::new(&view, &endpoint, limits); + let response = adapter.protocol_error( + ProtocolInteraction::Observe, + &ProtocolError { + code: ProtocolErrorCode::Internal, + message: "/private/backend/secret".repeat(1000), + }, + ); + assert!(response.body().len() < 512); + assert!( + !std::str::from_utf8(response.body()) + .unwrap() + .contains("secret") + ); + assert_eq!(response.status(), StatusCode::INTERNAL_SERVER_ERROR); + } +} diff --git a/crates/wip-fs-server/src/lib.rs b/crates/wip-fs-server/src/lib.rs index dbf690e..0470423 100644 --- a/crates/wip-fs-server/src/lib.rs +++ b/crates/wip-fs-server/src/lib.rs @@ -7,12 +7,46 @@ //! - [`wip_http`] owns routes, JSON, envelopes, and status mapping; //! - this crate owns the filesystem capability boundary and the real HTTP listener. //! +//! # Extension boundary +//! +//! [`Projection`] composes a [`FilesystemBackend`], typed [`Composition`] and +//! [`ProjectionPolicy`]. External local Objects can wrap a [`BackendEntry`] and +//! implement generated interfaces without orphan-rule workarounds. The backend +//! retains a root-confined target and atomic validator/dispatch boundary; the +//! engine owns recursive observation and standard host typed validation. No +//! listener is needed to use [`ProjectionView`]. [`HttpAdapter`] applies the +//! existing standard HTTP binding to a deployment-established principal view. +//! +//! Backends and Operations are trusted code, responsible for truthful effects, +//! public-only metadata and ongoing execution-context bounds. Policies must be +//! stable within a view; changing descriptor visibility requires new reference +//! lifetimes. See the crate README for implementer security and migration details. +//! +//! ```no_run +//! use wip_fs_server::{default_projection, AccessMode, ResourceLimits}; +//! use wip_host::{ExecutionContext, ObjectProvider}; +//! use wip_protocol::ObserveRequest; +//! +//! let engine = default_projection(std::path::Path::new("./public"), +//! AccessMode::ReadOnly, ResourceLimits::default())?; +//! let view = engine.view(&()); +//! let context = ExecutionContext::new().with_resource_budget(128); +//! let tree = view.observe(&context, &ObserveRequest { path: "/".into(), depth: 1 })?; +//! assert_eq!(tree.object.name, ""); +//! let object = view.observe_object_with_context(&context, "/")?; +//! assert!(object.interfaces.is_empty()); +//! # Ok::<(), Box>(()) +//! ``` +//! //! The built-in HTTP listener is intentionally loopback-only. TLS termination, //! certificates, authentication, and externally visible schemes are independent //! deployment concerns rather than WIP transport profiles. #![deny(missing_docs)] +mod defaults; +mod engine; +mod http_adapter; mod interfaces; mod projection; mod server; @@ -21,7 +55,17 @@ use std::net::{IpAddr, Ipv4Addr, SocketAddr}; use std::path::PathBuf; use std::time::Duration; -pub use interfaces::{FileRead, FileWrite, SymlinkInspect}; +pub use defaults::{DefaultComposition, DefaultInterfaces, DefaultObject, default_projection}; +pub use engine::{ + BackendEntry, Composition, DefaultPolicy, EntryKind, FilesystemBackend, Projection, + ProjectionPolicy, ProjectionView, +}; +pub use http_adapter::HttpAdapter; +pub use interfaces::{ + FileRead, FileReadDefinition, FileWrite, FileWriteDefinition, SymlinkInspect, + SymlinkInspectDefinition, +}; +pub use projection::{LinuxBackend, LinuxEntry}; pub use server::{BoundServer, ServerBuildError, WipFsServer}; /// Filesystem capabilities published by regular files. diff --git a/crates/wip-fs-server/src/projection.rs b/crates/wip-fs-server/src/projection.rs index adb498d..4f2c7c0 100644 --- a/crates/wip-fs-server/src/projection.rs +++ b/crates/wip-fs-server/src/projection.rs @@ -3,54 +3,37 @@ use std::mem::MaybeUninit; use std::os::fd::{AsRawFd, OwnedFd}; use std::path::{Component, Path, PathBuf}; use std::sync::atomic::{AtomicUsize, Ordering}; -use std::sync::{Arc, Mutex}; +use std::sync::{Arc, Mutex, MutexGuard, TryLockError}; +use crate::engine::{BackendEntry, EntryKind, FilesystemBackend}; +use crate::interfaces::{FileRead, FileWrite, SymlinkInspect}; +use crate::{AccessMode, ResourceLimits}; 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, + EffectGuarantee, ExecutionContext, HostError, OperationAttempt, ProviderResult, Validator, + WipObject, }; use wip_interface as wip; -use wip_protocol::{ - FetchInterfaceRequest, FetchInterfaceResponse, ObjectObservation, ObserveRequest, - ObserveResponse, 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, +fn entry_kind(mode: u32) -> EntryKind { + match FileType::from_raw_mode(mode) { + FileType::Directory => EntryKind::Directory, + FileType::RegularFile => EntryKind::Regular, + FileType::Symlink => EntryKind::Symlink, + _ => EntryKind::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", - } +fn entry_description(kind: EntryKind) -> &'static str { + match kind { + EntryKind::Directory => "Filesystem directory", + EntryKind::Regular => "Regular filesystem file", + EntryKind::Symlink => "Symbolic link (not followed)", + EntryKind::Unsupported => "Unsupported filesystem entry", } } @@ -98,19 +81,62 @@ impl Drop for HandlePermit { #[derive(Debug)] struct EntryInner { fd: OwnedFd, + execution: Mutex<()>, _permit: HandlePermit, } -#[derive(Debug, Clone)] -struct Entry { +/// A pinned Linux filesystem entry with bounded request-time capabilities. +/// +/// Cloning retains the same open descriptor and handle permit. Use capabilities +/// within [`FilesystemBackend::execute`] for serialized validator checks and +/// dispatch; direct calls still enforce entry kind, access, and request limits. +#[derive(Clone)] +pub struct LinuxEntry { name: String, kind: EntryKind, validator: Option, initial_validator: Option>, inner: Arc, + context: ExecutionContext, + access: AccessMode, + limits: ResourceLimits, + injection: FailureInjection, } -impl Entry { +impl LinuxEntry { + /// Returns the access mode used to open this entry. + #[must_use] + pub fn access_mode(&self) -> AccessMode { + self.access + } + + /// Returns whether this entry supports regular-file reads. + #[must_use] + pub fn supports_file_read(&self) -> bool { + self.kind == EntryKind::Regular + } + + /// Returns whether this entry supports regular-file replacement. + #[must_use] + pub fn supports_file_write(&self) -> bool { + self.kind == EntryKind::Regular && self.access == AccessMode::ReadWrite + } + + /// Returns whether this entry supports inspecting a symbolic link itself. + #[must_use] + pub fn supports_symlink_inspect(&self) -> bool { + self.kind == EntryKind::Symlink + } + + fn refresh_validator(&mut self) -> Result<(), ProjectionError> { + let stat = fstat(self.fd()).map_err(|_| ProjectionError::Internal)?; + let validator = matches!(self.kind, EntryKind::Regular | EntryKind::Symlink) + .then(|| validator_from_stat(&stat)); + self.initial_validator = validator.clone(); + self.validator = validator.map(Validator::new); + Ok(()) + } + fn fd(&self) -> &OwnedFd { &self.inner.fd } @@ -124,34 +150,23 @@ enum ProjectionError { } impl ProjectionError { - fn protocol(self) -> ProtocolError { + fn host(self) -> ProviderResult { 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(), - }, + Self::NotFound => ProviderResult::NotFound, + error => ProviderResult::Failure(error.host_error()), } } - fn host(self) -> ProviderResult<()> { + fn host_error(self) -> HostError { match self { - Self::NotFound => ProviderResult::NotFound, - Self::Resource => ProviderResult::Failure(HostError::resource_limit( + Self::Resource => HostError::resource_limit( "filesystem projection resource limit exceeded", EffectGuarantee::NotCommitted, - )), - Self::Internal => ProviderResult::Failure(HostError::internal( + ), + Self::NotFound | Self::Internal => HostError::internal( "filesystem projection failed", EffectGuarantee::NotCommitted, - )), + ), } } } @@ -163,42 +178,29 @@ enum FailureInjection { AfterTruncate, } -/// Request-time filesystem provider shared by the transport adapter. -pub(crate) struct Projection { +/// Linux filesystem backend using pinned, no-follow handles. +/// +/// Regular-file execution holds an advisory exclusive `flock` across validator +/// refresh, dispatch, and post-operation observation. Serialization only covers +/// actors that participate in this locking protocol; external writers can still +/// modify the inode without taking the lock. Symlinks are inspected, not followed. +pub struct LinuxBackend { root: OwnedFd, - registry: InterfaceRegistry, - read_interface: RegisteredInterface, - symlink_interface: RegisteredInterface, - write_interface: RegisteredInterface, access: AccessMode, limits: ResourceLimits, handles: Arc, injection: Mutex, } -impl Projection { - pub(crate) fn open( - root: &std::path::Path, +impl LinuxBackend { + /// Opens and pins a root directory without following symlink components. + pub fn open( + root: &Path, access: AccessMode, limits: ResourceLimits, ) -> Result { - let root = open_root(root)?; - 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, + root: open_root(root)?, access, limits, handles: Arc::new(HandleCounter { @@ -209,148 +211,16 @@ impl Projection { }) } - pub(crate) fn effective_root(&self) -> Result { + /// Returns the current filesystem path of the pinned root directory. + pub fn effective_root(&self) -> Result { std::fs::read_link(format!("/proc/self/fd/{}", self.root.as_raw_fd())) } - pub(crate) fn fetch_interface( - &self, - request: &FetchInterfaceRequest, - ) -> Result { - self.registry - .fetch_interface(&request.interface)? - .protocol_response() - } - - pub(crate) fn observe( + fn resolve_entry( &self, context: &ExecutionContext, - request: &ObserveRequest, - ) -> Result { - request.validate().map_err(|_| ProtocolError { - code: ProtocolErrorCode::InvalidRequest, - message: "invalid observe request".into(), - })?; - if request.depth > self.limits.max_observation_depth { - return Err(ProjectionError::Resource.protocol()); - } - - let root = self - .resolve(context, &request.path) - .map_err(ProjectionError::protocol)?; - let mut node_count = 0_u32; - let response = self.observe_entry(context, root, 0, request.depth, &mut node_count)?; - request - .validate_response(&response) - .map_err(|_| ProtocolError { - code: ProtocolErrorCode::Internal, - message: "Host produced an invalid observation response".into(), - })?; - Ok(response) - } - - fn observe_entry( - &self, - context: &ExecutionContext, - entry: Entry, - depth: u32, - maximum_depth: u32, - node_count: &mut u32, - ) -> Result { - check_context(context).map_err(ProjectionError::protocol)?; - *node_count = node_count.saturating_add(1); - if *node_count > self.limits.max_observation_nodes { - return Err(ProjectionError::Resource.protocol()); - } - - let object = self - .publication(entry.clone(), context.clone()) - .protocol_object(); - if depth == maximum_depth { - return Ok(ObjectObservation { - object, - children: None, - }); - } - - let mut children = Vec::new(); - if entry.kind == EntryKind::Directory { - let directory_fd = reopen(entry.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, entry.fd(), name) { - Ok(child) => child, - Err(ProjectionError::NotFound) => continue, - Err(error) => return Err(error.protocol()), - }; - children.push(self.observe_entry( - context, - child, - depth + 1, - maximum_depth, - node_count, - )?); - } - } - - Ok(ObjectObservation { - object, - children: Some(children), - }) - } - - 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 { + path: &str, + ) -> Result { check_context(context)?; wip_protocol::validate_path(path).map_err(|_| ProjectionError::NotFound)?; if path == "/" { @@ -372,7 +242,7 @@ impl Projection { ) .map_err(classify_lookup)?; let stat = fstat(&probe).map_err(|_| ProjectionError::Internal)?; - let kind = EntryKind::from_mode(stat.st_mode); + let kind = entry_kind(stat.st_mode); if segments.peek().is_some() { if kind != EntryKind::Directory { return Err(ProjectionError::NotFound); @@ -391,7 +261,7 @@ impl Projection { context: &ExecutionContext, parent: &OwnedFd, name: &str, - ) -> Result { + ) -> Result { check_context(context)?; let probe = openat( parent, @@ -401,7 +271,7 @@ impl Projection { ) .map_err(classify_lookup)?; let stat = fstat(&probe).map_err(|_| ProjectionError::Internal)?; - let kind = EntryKind::from_mode(stat.st_mode); + let kind = entry_kind(stat.st_mode); let fd = self.reopen_final(probe, kind)?; self.entry(context, name.to_owned(), fd) } @@ -426,20 +296,28 @@ impl Projection { context: &ExecutionContext, name: String, fd: OwnedFd, - ) -> Result { + ) -> 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 kind = entry_kind(stat.st_mode); let validator = matches!(kind, EntryKind::Regular | EntryKind::Symlink) .then(|| validator_from_stat(&stat)); - Ok(Entry { + Ok(LinuxEntry { name, kind, initial_validator: validator.clone(), validator: validator.map(Validator::new), + context: context.clone(), + access: self.access, + limits: self.limits, + injection: *self + .injection + .lock() + .unwrap_or_else(|error| error.into_inner()), inner: Arc::new(EntryInner { fd, + execution: Mutex::new(()), _permit: permit, }), }) @@ -454,54 +332,77 @@ impl Projection { } } -impl ObjectProvider for Projection { - type Object = FilesystemObject; +impl BackendEntry for LinuxEntry { + fn kind(&self) -> EntryKind { + self.kind + } +} - 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)) +impl FilesystemBackend for LinuxBackend { + type Entry = LinuxEntry; + + fn resolve(&self, context: &ExecutionContext, path: &str) -> ProviderResult { + match self.resolve_entry(context, path) { + Ok(entry) => ProviderResult::Found(entry), + Err(error) => error.host(), + } } - fn execute_operation( + fn children( &self, context: &ExecutionContext, - path: &str, - operation: impl FnOnce(&PublishedObject) -> OperationAttempt, + entry: &LinuxEntry, + maximum: usize, + ) -> Result, HostError> { + check_context(context).map_err(resource_host)?; + if entry.kind != EntryKind::Directory { + return Ok(Vec::new()); + } + let directory_fd = reopen(entry.fd(), OFlags::RDONLY | OFlags::DIRECTORY) + .map_err(ProjectionError::host_error)?; + let mut buffer = [MaybeUninit::::uninit(); 8192]; + let mut directory = RawDir::new(directory_fd, &mut buffer); + let mut children = Vec::new(); + while let Some(item) = directory.next() { + check_context(context).map_err(resource_host)?; + let item = item.map_err(|_| internal_not_committed())?; + 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, entry.fd(), name) { + Ok(child) => child, + Err(ProjectionError::NotFound) => continue, + Err(error) => return Err(error.host_error()), + }; + if children.len() >= maximum { + return Err(resource_not_committed()); + } + children.push(child); + } + Ok(children) + } + + fn execute( + &self, + context: &ExecutionContext, + mut entry: LinuxEntry, + callback: impl FnOnce(LinuxEntry) -> OperationAttempt, ) -> ProviderResult { - 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 { - match lock_with_context(entry.fd(), context, self.limits.request_timeout) { - Ok(()) => {} + if let Err(error) = check_context(context) { + return error.host(); + } + let lock_entry = Arc::clone(&entry.inner); + let _lock = if entry.kind == EntryKind::Regular { + match lock_with_context(&lock_entry, context, self.limits.request_timeout) { + Ok(lock) => Some(lock), Err(ProjectionError::Resource) => { return ProviderResult::Failure(HostError::resource_limit( "filesystem lock deadline exceeded", @@ -515,59 +416,62 @@ impl ObjectProvider for Projection { )); } } + } else { + None + }; + // Both snapshots must describe the state under the lock, not at resolve time. + if entry.refresh_validator().is_err() { + return ProviderResult::Failure(internal_not_committed()); } - let publication = self.publication(entry.clone(), context.clone()); - let mut attempt = operation(&publication); + entry.context = context.clone(); + let mut attempt = callback(entry.clone()); 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); + if entry.refresh_validator().is_err() { + return ProviderResult::Failure(HostError::internal( + "filesystem state could not be observed after dispatch", + EffectGuarantee::Unknown, + )); + } + result.validator = entry.initial_validator.clone(); } ProviderResult::Found(attempt) } } -pub(crate) struct FilesystemObject { - entry: Entry, - context: ExecutionContext, - limits: ResourceLimits, - injection: FailureInjection, +// Unlock before releasing clone serialization, including on refresh/dispatch failure. +struct EntryLock<'a> { + inner: &'a EntryInner, + _guard: MutexGuard<'a, ()>, } -impl WipObject for FilesystemObject { +impl Drop for EntryLock<'_> { + fn drop(&mut self) { + let _ = flock(&self.inner.fd, FlockOperation::Unlock); + } +} + +impl WipObject for LinuxEntry { fn name(&self) -> &str { - &self.entry.name + &self.name } fn description(&self) -> Option<&str> { - Some(self.entry.kind.description()) + Some(entry_description(self.kind)) } fn validator(&self) -> Option<&Validator> { - self.entry.validator.as_ref() - } - - fn requires_validator(&self, _: &wip_host::InterfaceRef, operation: &str) -> bool { - operation == "replace" + self.validator.as_ref() } } -impl FileRead for FilesystemObject { +impl FileRead for LinuxEntry { fn read(&self, offset: Option, length: Option) -> wip::OperationResult { check_context(&self.context).map_err(resource_host)?; + if !self.supports_file_read() { + return Err(internal_not_committed()); + } let offset = offset.unwrap_or(0); - let stat = fstat(self.entry.fd()).map_err(|_| internal_not_committed())?; + let stat = fstat(self.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 { @@ -583,7 +487,7 @@ impl FileRead for FilesystemObject { 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) + let count = pread(self.fd(), &mut bytes[read..], offset + read as u64) .map_err(|_| internal_not_committed())?; if count == 0 { break; @@ -595,24 +499,28 @@ impl FileRead for FilesystemObject { } } -impl SymlinkInspect for FilesystemObject { +impl SymlinkInspect for LinuxEntry { 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())?; + if !self.supports_symlink_inspect() { + return Err(internal_not_committed()); + } + let target = readlinkat(self.fd(), "", Vec::new()).map_err(|_| internal_not_committed())?; Ok(wip::Bytes(target.as_bytes().to_vec())) } } -impl FileWrite for FilesystemObject { +impl FileWrite for LinuxEntry { fn replace(&self, contents: wip::Bytes) -> wip::OperationResult<()> { check_context(&self.context).map_err(resource_host)?; + if !self.supports_file_write() { + return Err(internal_not_committed()); + } 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()) - { + let current = fstat(self.fd()).map_err(|_| internal_not_committed())?; + if self.initial_validator.as_deref() != Some(validator_from_stat(¤t).as_slice()) { return Err(HostError::internal( "filesystem object changed before write", EffectGuarantee::NotCommitted, @@ -621,7 +529,7 @@ impl FileWrite for FilesystemObject { if self.injection == FailureInjection::BeforeWrite { return Err(internal_not_committed()); } - ftruncate(self.entry.fd(), 0).map_err(|_| internal_not_committed())?; + ftruncate(self.fd(), 0).map_err(|_| internal_not_committed())?; if self.injection == FailureInjection::AfterTruncate { return Err(HostError::internal( "injected failure after write began", @@ -637,7 +545,7 @@ impl FileWrite for FilesystemObject { )); } let count = - pwrite(self.entry.fd(), &contents.0[written..], written as u64).map_err(|_| { + pwrite(self.fd(), &contents.0[written..], written as u64).map_err(|_| { HostError::internal("filesystem write failed", EffectGuarantee::Unknown) })?; if count == 0 { @@ -648,7 +556,7 @@ impl FileWrite for FilesystemObject { } written += count; } - rustix::fs::fdatasync(self.entry.fd()).map_err(|_| { + rustix::fs::fdatasync(self.fd()).map_err(|_| { HostError::internal( "filesystem synchronization failed", EffectGuarantee::Unknown, @@ -686,7 +594,7 @@ fn open_root(path: &Path) -> Result { ) .map_err(std::io::Error::from)?; let stat = fstat(&probe).map_err(std::io::Error::from)?; - if EntryKind::from_mode(stat.st_mode) != EntryKind::Directory { + if entry_kind(stat.st_mode) != EntryKind::Directory { return Err(std::io::Error::new( std::io::ErrorKind::NotADirectory, "filesystem root contains a symlink or non-directory component", @@ -698,11 +606,11 @@ fn open_root(path: &Path) -> Result { Ok(current) } -fn lock_with_context( - fd: &OwnedFd, +fn lock_with_context<'a>( + inner: &'a EntryInner, context: &ExecutionContext, maximum_wait: std::time::Duration, -) -> Result<(), ProjectionError> { +) -> Result, ProjectionError> { let now = std::time::Instant::now(); let fallback_deadline = now.checked_add(maximum_wait).unwrap_or(now); let deadline = context.deadline().map_or(fallback_deadline, |configured| { @@ -713,9 +621,24 @@ fn lock_with_context( if std::time::Instant::now() >= deadline { return Err(ProjectionError::Resource); } - match flock(fd, FlockOperation::NonBlockingLockExclusive) { - Ok(()) => return Ok(()), + // flock treats clones' shared open-file description as a single owner. + let guard = match inner.execution.try_lock() { + Ok(guard) => guard, + Err(TryLockError::Poisoned(error)) => error.into_inner(), + Err(TryLockError::WouldBlock) => { + std::thread::sleep(std::time::Duration::from_millis(1)); + continue; + } + }; + match flock(&inner.fd, FlockOperation::NonBlockingLockExclusive) { + Ok(()) => { + return Ok(EntryLock { + inner, + _guard: guard, + }); + } Err(rustix::io::Errno::WOULDBLOCK) => { + drop(guard); std::thread::sleep(std::time::Duration::from_millis(1)); } Err(_) => return Err(ProjectionError::Internal), @@ -767,7 +690,7 @@ fn context_expired(context: &ExecutionContext) -> bool { } fn check_context(context: &ExecutionContext) -> Result<(), ProjectionError> { - if context_expired(context) { + if context_expired(context) || context.resource_budget() == Some(0) { Err(ProjectionError::Resource) } else { Ok(()) @@ -804,19 +727,24 @@ mod tests { use tempfile::tempdir; use wip_host::{ - CallErrorKind, CallRequest, ExecutionContext, InterfaceTarget, ObjectProvider, Target, + CallErrorKind, CallRequest, ExecutionContext, InterfaceTarget, ObjectProvider, + OperationAttempt, ProviderResult, Target, WipObject, }; use wip_protocol::{ObserveRequest, ProtocolErrorCode, Value}; - use super::{FailureInjection, FlockOperation, Projection}; - use crate::{AccessMode, ResourceLimits}; + use super::{FailureInjection, FlockOperation, LinuxBackend, LinuxEntry}; + use crate::engine::{DefaultPolicy, FilesystemBackend, Projection}; + use crate::interfaces::{FileRead, FileWrite}; + use crate::{AccessMode, DefaultComposition, ResourceLimits}; - fn projection(root: &std::path::Path, access: AccessMode) -> Projection { - Projection::open(root, access, ResourceLimits::default()).expect("open projection") + type DefaultProjection = Projection; + + fn projection(root: &std::path::Path, access: AccessMode) -> DefaultProjection { + crate::default_projection(root, access, ResourceLimits::default()).expect("open projection") } fn call( - projection: &Projection, + projection: &DefaultProjection, path: &str, validator: Option>, interface: String, @@ -837,6 +765,245 @@ mod tests { }) } + fn resolved(backend: &LinuxBackend, context: &ExecutionContext, path: &str) -> LinuxEntry { + match backend.resolve(context, path) { + ProviderResult::Found(entry) => entry, + ProviderResult::NotFound => panic!("entry should resolve"), + ProviderResult::Failure(error) => panic!("resolution failed: {error:?}"), + } + } + + #[test] + fn execution_refreshes_both_validators_under_lock_and_returns_post_write_validator() { + let temporary = tempdir().expect("temporary root"); + let path = temporary.path().join("file"); + fs::write(&path, b"old").expect("write file"); + let backend = LinuxBackend::open( + temporary.path(), + AccessMode::ReadWrite, + ResourceLimits::default(), + ) + .expect("open backend"); + let context = ExecutionContext::new(); + let entry = resolved(&backend, &context, "/file"); + let original = entry.initial_validator.clone(); + let pinned = entry.clone(); + // Simulate a cooperating writer committing between resolution and execution. + fs::write(&path, b"changed before lock acquisition").expect("external mutation"); + let lock_probe = fs::OpenOptions::new() + .read(true) + .write(true) + .open(&path) + .expect("open independent lock probe"); + let mut callback_validator = None; + let attempt = backend.execute(&context, entry, |entry| { + assert!(std::sync::Arc::ptr_eq(&entry.inner, &pinned.inner)); + assert_ne!(entry.initial_validator, original); + assert_eq!( + entry + .validator() + .map(|validator| validator.as_bytes().to_vec()), + entry.initial_validator, + ); + callback_validator = entry.initial_validator.clone(); + assert_eq!( + rustix::fs::flock(&lock_probe, FlockOperation::NonBlockingLockExclusive), + Err(rustix::io::Errno::WOULDBLOCK), + ); + entry + .replace(wip_interface::Bytes(b"new".to_vec())) + .expect("write uses refreshed initial validator"); + OperationAttempt::Started(Ok(wip_host::CallResult { + result: Value::Bytes(Vec::new()), + validator: None, + })) + }); + let ProviderResult::Found(OperationAttempt::Started(Ok(result))) = attempt else { + panic!("execution should succeed: {attempt:?}"); + }; + assert_ne!(result.validator, callback_validator); + assert_eq!( + result.validator, + resolved(&backend, &context, "/file").initial_validator, + ); + assert_eq!(fs::read(&path).expect("read written file"), b"new"); + // The retained clone must not keep the operation's flock held. + rustix::fs::flock(&lock_probe, FlockOperation::NonBlockingLockExclusive) + .expect("execution released lock"); + rustix::fs::flock(&lock_probe, FlockOperation::Unlock).expect("unlock probe"); + } + + #[test] + fn execution_uses_the_pinned_entry_after_path_replacement() { + let temporary = tempdir().expect("temporary root"); + let path = temporary.path().join("file"); + fs::write(&path, b"old").expect("write file"); + let backend = LinuxBackend::open( + temporary.path(), + AccessMode::ReadWrite, + ResourceLimits::default(), + ) + .expect("open backend"); + let context = ExecutionContext::new(); + let entry = resolved(&backend, &context, "/file"); + let pinned = entry.clone(); + fs::remove_file(&path).expect("unlink pinned entry"); + fs::write(&path, b"replacement").expect("replace path"); + let attempt = backend.execute(&context, entry, |entry| { + assert!(std::sync::Arc::ptr_eq(&entry.inner, &pinned.inner)); + entry + .replace(wip_interface::Bytes(b"pinned write".to_vec())) + .expect("write pinned inode"); + OperationAttempt::Started(Ok(wip_host::CallResult { + result: Value::Bytes(Vec::new()), + validator: None, + })) + }); + assert!(matches!( + attempt, + ProviderResult::Found(OperationAttempt::Started(Ok(_))) + )); + assert_eq!(fs::read(path).expect("read replacement"), b"replacement"); + assert_eq!( + pinned.read(None, None).expect("read pinned inode").0, + b"pinned write" + ); + } + + #[test] + fn rejected_execution_releases_lock_even_when_callback_retains_entry() { + let temporary = tempdir().expect("temporary root"); + let path = temporary.path().join("file"); + fs::write(&path, b"old").expect("write file"); + let backend = LinuxBackend::open( + temporary.path(), + AccessMode::ReadOnly, + ResourceLimits::default(), + ) + .expect("open backend"); + let context = ExecutionContext::new(); + let entry = resolved(&backend, &context, "/file"); + let mut retained = None; + let attempt = backend.execute(&context, entry, |entry| { + retained = Some(entry); + OperationAttempt::reject(CallErrorKind::ValidatorMismatch) + }); + assert!(matches!( + attempt, + ProviderResult::Found(OperationAttempt::NotStarted(_)) + )); + let lock_probe = fs::File::open(path).expect("open independent lock probe"); + rustix::fs::flock(&lock_probe, FlockOperation::NonBlockingLockExclusive) + .expect("rejected execution released lock"); + rustix::fs::flock(&lock_probe, FlockOperation::Unlock).expect("unlock probe"); + assert!(retained.is_some()); + } + + #[test] + fn cloned_entries_cannot_bypass_execution_lock_and_wait_is_bounded() { + let temporary = tempdir().expect("temporary root"); + fs::write(temporary.path().join("file"), b"old").expect("write file"); + let backend = LinuxBackend::open( + temporary.path(), + AccessMode::ReadOnly, + ResourceLimits { + request_timeout: std::time::Duration::from_millis(20), + ..ResourceLimits::default() + }, + ) + .expect("open backend"); + let context = ExecutionContext::new(); + let entry = resolved(&backend, &context, "/file"); + let cloned = entry.clone(); + let attempt = backend.execute(&context, entry, |entry| { + let started = std::time::Instant::now(); + let nested = backend.execute(&context, entry, |_| { + panic!("shared open-file description must not bypass execution lock"); + }); + let ProviderResult::Failure(error) = nested else { + panic!("nested clone execution must time out: {nested:?}"); + }; + assert_eq!(error.kind(), wip_host::HostErrorKind::ResourceLimitExceeded); + assert_eq!( + error.effect_guarantee(), + wip_host::EffectGuarantee::NotCommitted + ); + assert!(started.elapsed() < std::time::Duration::from_secs(1)); + OperationAttempt::reject(CallErrorKind::ValidatorMismatch) + }); + assert!(matches!(attempt, ProviderResult::Found(_))); + assert!(matches!( + backend.execute(&context, cloned, |_| { + OperationAttempt::reject(CallErrorKind::ValidatorMismatch) + }), + ProviderResult::Found(_) + )); + } + + #[test] + fn children_budget_fails_completely_and_drops_temporary_handles() { + let temporary = tempdir().expect("temporary root"); + fs::write(temporary.path().join("a"), b"a").expect("write file"); + fs::write(temporary.path().join("b"), b"b").expect("write file"); + let backend = LinuxBackend::open( + temporary.path(), + AccessMode::ReadOnly, + ResourceLimits { + max_open_handles: 4, + ..ResourceLimits::default() + }, + ) + .expect("open backend"); + let context = ExecutionContext::new(); + let root = resolved(&backend, &context, "/"); + assert!(backend.children(&context, &root, 1).is_err()); + assert_eq!(backend.handles.current.load(super::Ordering::Acquire), 1); + let children = backend + .children(&context, &root, 2) + .expect("complete children"); + assert_eq!(children.len(), 2); + let cloned = children[0].clone(); + assert_eq!(backend.handles.current.load(super::Ordering::Acquire), 3); + drop(children); + assert_eq!(backend.handles.current.load(super::Ordering::Acquire), 2); + drop(cloned); + assert_eq!(backend.handles.current.load(super::Ordering::Acquire), 1); + } + + #[test] + fn direct_capabilities_enforce_access_limits_and_execution_context() { + let temporary = tempdir().expect("temporary root"); + let path = temporary.path().join("file"); + fs::write(&path, b"old").expect("write file"); + let backend = LinuxBackend::open( + temporary.path(), + AccessMode::ReadOnly, + ResourceLimits { + max_file_read_bytes: 2, + ..ResourceLimits::default() + }, + ) + .expect("open backend"); + let context = ExecutionContext::new(); + let entry = resolved(&backend, &context, "/file"); + assert_eq!(entry.access_mode(), AccessMode::ReadOnly); + assert!(entry.supports_file_read()); + assert!(!entry.supports_file_write()); + assert!(!entry.supports_symlink_inspect()); + assert!(entry.read(None, None).is_err()); + assert_eq!(entry.read(None, Some(2)).expect("bounded read").0, b"ol"); + assert!(entry.replace(wip_interface::Bytes(Vec::new())).is_err()); + assert_eq!(fs::read(&path).expect("read unchanged file"), b"old"); + let cancellation = wip_host::CancellationToken::new(); + cancellation.cancel(); + let attempt = backend.execute( + &ExecutionContext::new().with_cancellation(cancellation), + entry, + |_| panic!("cancelled execution must not invoke callback"), + ); + assert!(matches!(attempt, ProviderResult::Failure(_))); + } + #[test] fn projects_entry_kinds_without_following_symlinks() { let temporary = tempdir().expect("temporary root"); @@ -1008,14 +1175,18 @@ mod tests { ) }; - projection.set_injection(FailureInjection::BeforeWrite); + projection + .backend() + .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); + projection + .backend() + .set_injection(FailureInjection::AfterTruncate); assert_eq!( request().expect_err("post-truncate failure").kind(), CallErrorKind::OperationOutcomeUnknown @@ -1114,7 +1285,7 @@ mod tests { let linked = temporary.path().join("linked"); std::os::unix::fs::symlink(&actual, &linked).expect("create root symlink"); assert!( - Projection::open(&linked, AccessMode::ReadOnly, ResourceLimits::default()).is_err() + LinuxBackend::open(&linked, AccessMode::ReadOnly, ResourceLimits::default()).is_err() ); } @@ -1134,7 +1305,7 @@ mod tests { request_timeout: std::time::Duration::from_millis(20), ..ResourceLimits::default() }; - let projection = Projection::open(temporary.path(), AccessMode::ReadOnly, limits) + let projection = crate::default_projection(temporary.path(), AccessMode::ReadOnly, limits) .expect("open projection"); let observed = projection.observe_object("/file").expect("observe file"); let started = std::time::Instant::now(); diff --git a/crates/wip-fs-server/src/server.rs b/crates/wip-fs-server/src/server.rs index 1294e3f..77edc50 100644 --- a/crates/wip-fs-server/src/server.rs +++ b/crates/wip-fs-server/src/server.rs @@ -5,26 +5,20 @@ use std::sync::atomic::{AtomicBool, Ordering}; use std::thread; use std::time::{Duration, Instant}; -use http::header::ALLOW; use http::{Request, Response, StatusCode}; -use wip_host::{ExecutionContext, ObjectProvider}; -use wip_http::{ - CodecError, EncodeHttpError, Endpoint, RequestMetadataError, Route, - decode_call_operation_metadata, decode_fetch_interface_request, decode_observe_request, - encode_call_operation_response, encode_fetch_interface_response, encode_observe_response, - encode_protocol_error_response, validate_http_request, -}; -use wip_protocol::{ProtocolError, ProtocolErrorCode, ProtocolInteraction}; +use wip_host::ExecutionContext; +use wip_http::Endpoint; -use crate::projection::Projection; -use crate::{ConfigError, ServerConfig}; +use crate::engine::Projection; +use crate::http_adapter::HttpAdapter; +use crate::{ConfigError, DefaultComposition, DefaultPolicy, LinuxBackend, ServerConfig}; -/// Filesystem projection and framework-neutral standard WIP request adapter. +/// Default Linux filesystem projection and bounded loopback HTTP listener. pub struct WipFsServer { config: ServerConfig, endpoint: Endpoint, effective_root: std::path::PathBuf, - projection: Arc, + projection: Arc>, } impl WipFsServer { @@ -32,8 +26,8 @@ impl WipFsServer { pub fn new(config: ServerConfig) -> Result { config.validate()?; let endpoint = config.endpoint()?; - let projection = Projection::open(&config.root, config.access, config.limits)?; - let effective_root = projection.effective_root()?; + let projection = crate::default_projection(&config.root, config.access, config.limits)?; + let effective_root = projection.backend().effective_root()?; Ok(Self { config, endpoint, @@ -67,25 +61,12 @@ impl WipFsServer { request: Request>, deadline: Instant, ) -> Response> { - let Some(route) = self.route(request.uri()) else { - return empty_response(StatusCode::NOT_FOUND); - }; - if let Err(error) = validate_http_request(&self.endpoint, route, &request) { - return self.metadata_error(error); - } let context = ExecutionContext::new() .with_deadline(deadline) .with_resource_budget(u64::from(self.config.limits.max_observation_nodes)); - let response = match route { - Route::Observe => self.observe(request.body(), &context), - Route::FetchInterface => self.fetch_interface(request.body()), - Route::CallOperation => self.call_operation(request.body(), &context), - }; - if Instant::now() >= deadline { - empty_response(StatusCode::GATEWAY_TIMEOUT) - } else { - response - } + let view = self.projection.view(&()); + HttpAdapter::new(&view, &self.endpoint, self.config.limits) + .handle_http_with_context(request, &context) } /// Binds the built-in bounded loopback HTTP listener. @@ -108,171 +89,6 @@ impl WipFsServer { listener: Arc::new(listener), }) } - - fn route(&self, uri: &http::Uri) -> Option { - if uri.query().is_some() { - return None; - } - [Route::Observe, Route::FetchInterface, Route::CallOperation] - .into_iter() - .find(|route| self.endpoint.route(*route).path() == uri.path()) - } - - fn observe(&self, body: &[u8], context: &ExecutionContext) -> Response> { - let request = match decode_observe_request(body, self.config.limits.wire) { - Ok(request) => request, - Err(error) => return self.codec_error(ProtocolInteraction::Observe, error), - }; - match self.projection.observe(context, &request) { - Ok(tree) => encode_observe_response(&request, &tree, self.config.limits.wire) - .unwrap_or_else(|error| { - self.read_encode_error(ProtocolInteraction::Observe, error) - }), - Err(error) => self.protocol_error(ProtocolInteraction::Observe, &error), - } - } - - 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), - } - } - - 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 may_commit = metadata.operation == "replace"; - 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 { - let error = ProtocolError { - code: if may_commit { - ProtocolErrorCode::OperationOutcomeUnknown - } else { - ProtocolErrorCode::Internal - }, - message: "operation completed without response metadata".into(), - }; - return self.protocol_error(ProtocolInteraction::CallOperation, &error); - }; - encode_call_operation_response( - &request, - &descriptor, - &result, - self.config.limits.wire, - ) - .unwrap_or_else(|encode_error| { - if matches!( - &encode_error, - EncodeHttpError::Codec(codec) if is_transport_limit(codec) - ) { - return empty_response(StatusCode::INTERNAL_SERVER_ERROR); - } - if !may_commit { - return self - .read_encode_error(ProtocolInteraction::CallOperation, encode_error); - } - // Replacement may already be committed, so a response - // encoding failure cannot carry a non-commit guarantee. - let error = ProtocolError { - code: ProtocolErrorCode::OperationOutcomeUnknown, - message: "operation completed but its response could not be encoded".into(), - }; - self.protocol_error(ProtocolInteraction::CallOperation, &error) - }) - } - Err(error) => self.protocol_error(ProtocolInteraction::CallOperation, &error), - } - } - - fn metadata_error(&self, error: RequestMetadataError) -> Response> { - match error { - RequestMetadataError::NotFound => empty_response(StatusCode::NOT_FOUND), - error @ RequestMetadataError::MethodNotAllowed => Response::builder() - .status(error.status()) - .header( - ALLOW, - error - .allow() - .expect("method-not-allowed metadata has an Allow value"), - ) - .body(Vec::new()) - .expect("static method response is valid"), - 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 is_transport_limit(&error) { - empty_response(StatusCode::PAYLOAD_TOO_LARGE) - } else { - self.protocol_error( - interaction, - &ProtocolError { - code: ProtocolErrorCode::InvalidRequest, - message: "malformed WIP request".into(), - }, - ) - } - } - - fn read_encode_error( - &self, - interaction: ProtocolInteraction, - error: EncodeHttpError, - ) -> Response> { - match error { - EncodeHttpError::Codec(codec) if is_transport_limit(&codec) => { - empty_response(StatusCode::INTERNAL_SERVER_ERROR) - } - _ => self.protocol_error(interaction, &internal_error()), - } - } - - fn protocol_error( - &self, - interaction: ProtocolInteraction, - error: &ProtocolError, - ) -> Response> { - encode_protocol_error_response(interaction, error, self.config.limits.wire) - .unwrap_or_else(|_| empty_response(StatusCode::INTERNAL_SERVER_ERROR)) - } } /// Bound real HTTP listener for a [`WipFsServer`]. @@ -560,20 +376,6 @@ fn write_all_with_deadline( stream.flush() } -fn is_transport_limit(error: &CodecError) -> bool { - matches!( - error, - CodecError::BodyTooLarge { .. } | CodecError::NestingTooDeep { .. } - ) -} - -fn internal_error() -> ProtocolError { - ProtocolError { - code: ProtocolErrorCode::Internal, - message: "filesystem server failed".into(), - } -} - fn empty_response(status: StatusCode) -> Response> { Response::builder() .status(status)