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)