diff --git a/AGENTS.md b/AGENTS.md index 2ace06c7..08ef2740 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -104,12 +104,18 @@ discussion in the pull request: control story in full. - **Every call has a deadline.** It cannot be disabled. A call with no deadline reintroduces the hang that motivated the project. -- **A confidential message goes to a loaded, hash-verified module or to - nobody.** Only a module whose artifact the host hashed against a digest an +- **A confidential call goes to a loaded, hash-verified module, an explicitly + scoped SDK-created local host callback, or nobody.** Only a module whose artifact the host hashed against a digest an operator asserted — a `modules.toml` beside the file, or one compiled into the host and checked against the release manifest before extraction — may receive - one; a peer reached across a transport never can, - by design rather than by omission. The message is never fanned out to a + one. The sole host exception is `ModuleHost::prepare_host_callbacks` plus + `load_file_with_host_callbacks`: a private SDK-created MemoryTransport + connection, exact verified module unique sender, exact host unique destination, + path, interface and member, all bound before initialization. This capability + is not artifact attestation; aliases, other instances, ordinary connections and + custom/remote transports never inherit it. Revocation and detach prevent new + route and guarded user-dispatch admission, not cessation of already running + provider work. Generic module loads grant no host callback authority. The message is never fanned out to a subscriber, never printed by `monitor`, and never carried by a signal. This is admission control, not isolation — a loaded module is already inside the trust boundary and could read host memory directly; what the rule buys is that the diff --git a/crates/tinybus/Cargo.toml b/crates/tinybus/Cargo.toml index 78edb049..b567ed1e 100644 --- a/crates/tinybus/Cargo.toml +++ b/crates/tinybus/Cargo.toml @@ -130,3 +130,8 @@ required-features = ["modules"] name = "module_context_eager" crate-type = ["cdylib"] required-features = ["modules"] + +[[example]] +name = "module_host_callback" +crate-type = ["cdylib"] +required-features = ["modules"] diff --git a/crates/tinybus/examples/module_context.rs b/crates/tinybus/examples/module_context.rs index 16fcb345..a91c1717 100644 --- a/crates/tinybus/examples/module_context.rs +++ b/crates/tinybus/examples/module_context.rs @@ -12,7 +12,11 @@ impl Interface for ContextService { "org.example.Context".parse().unwrap() } fn members(&self) -> Vec { - vec!["Inspect".parse().unwrap(), "Callback".parse().unwrap()] + vec![ + "Inspect".parse().unwrap(), + "Callback".parse().unwrap(), + "ConfidentialCallback".parse().unwrap(), + ] } async fn call(&self, _: &MemberName, _: Value) -> Result { Err(Error::failed("authenticated incoming context required")) @@ -31,12 +35,15 @@ impl Interface for ContextService { } match member.as_str() { "Inspect" => Ok(serde_json::json!(sender)), - "Callback" => { - let reply: String = self + "Callback" | "ConfidentialCallback" => { + let proxy = self .0 - .proxy(sender.as_str(), "/callback", "org.example.Callback")? - .call("Observe", ()) - .await?; + .proxy(sender.as_str(), "/callback", "org.example.Callback")?; + let reply: String = if member.as_str() == "ConfidentialCallback" { + proxy.call_confidential("Observe", ()).await? + } else { + proxy.call("Observe", ()).await? + }; Ok(serde_json::json!([sender, reply])) } _ => Err(Error::failed("unknown method")), @@ -55,6 +62,6 @@ tinybus_module::module_export! { setup = setup, worker_threads = 1, provides = ["org.example.Context"], - methods = ["Inspect", "Callback"], + methods = ["Inspect", "Callback", "ConfidentialCallback"], signals = [], requires = [], optional = [], lazy = true, } diff --git a/crates/tinybus/examples/module_host_callback.rs b/crates/tinybus/examples/module_host_callback.rs new file mode 100644 index 00000000..d4f0d63e --- /dev/null +++ b/crates/tinybus/examples/module_host_callback.rs @@ -0,0 +1,30 @@ +//! Actual eager native fixture calling its fixed host during setup before READY. +use serde::Deserialize; +use tinybus::{Connection, Error, Result}; + +#[derive(Deserialize)] +struct Config { + host: tinybus::BusName, + #[serde(default)] + fail: bool, +} +async fn setup(connection: Connection, config: Config) -> Result<()> { + async { + if config.fail { + return Err(Error::failed("fixture initialization rejected")); + } + let _: tinybus::BusName = connection + .proxy(config.host.as_str(), "/callback", "org.example.Callback")? + .call_confidential("Observe", ()) + .await?; + connection.request_name("org.example.EagerCallback").await + } + .await +} + +tinybus_module::module_export! { + setup = setup, config = Config, + worker_threads = 1, + provides = ["org.example.EagerCallback"], + methods = ["Observe"], signals = [], requires = [], optional = [], lazy = false, +} diff --git a/crates/tinybus/src/attest.rs b/crates/tinybus/src/attest.rs index 308600ee..d3d45cf3 100644 --- a/crates/tinybus/src/attest.rs +++ b/crates/tinybus/src/attest.rs @@ -6,8 +6,11 @@ //! a recovery phrase, a bearer token. The recipient of one of those is a //! **module loaded into the host's address space**, whose artifact the host //! verified before `dlopen`, or whose code was linked into the host. A -//! secret handed to such a module never crosses a transport, never reaches a -//! separate process, and never touches a socket. +//! secret handed to such a module never reaches a separate process or socket. +//! An explicit `ModuleHost` capability also permits exact confidential callbacks +//! from one verified loaded instance to an SDK-created local host connection. +//! That scoped exception is not artifact attestation and grants no permission +//! to aliases, ordinary connections, custom transports or other module instances. //! //! Everything else on the bus — services in their own processes, CLI clients, //! monitors — does not receive secrets. Not "receives them less safely": @@ -38,8 +41,8 @@ //! //! What attestation buys is therefore **admission control**, not isolation: //! only code whose hash an operator allowlisted is loaded at all, and only such -//! code is handed a secret through the bus. The bus's job is to refuse to be -//! the delivery mechanism for anything else. An integration whose compromise +//! code or its explicitly scoped local host callback is handed a secret through +//! the bus. The bus refuses all other recipient classes. An integration whose compromise //! must not reach the kernel's secrets belongs in a separate process — where it //! is, by this design, ineligible to receive them. //! diff --git a/crates/tinybus/src/broker.rs b/crates/tinybus/src/broker.rs index 897f52fb..ce4db009 100644 --- a/crates/tinybus/src/broker.rs +++ b/crates/tinybus/src/broker.rs @@ -43,7 +43,7 @@ pub const PEER_QUEUE_CAPACITY: usize = 256; /// Cheap to clone; every clone shares one routing table. #[derive(Clone)] pub struct Broker { - router: Arc>, + pub(crate) router: Arc>, id: String, // `Weak`, not `Arc`: the module host owns this broker, so a strong // reference back would form a cycle and leak both. Callers tolerate a @@ -212,7 +212,7 @@ impl Broker { /// Record that a loaded module's artifact matched the module allowlist. /// - /// This is the *only* way a peer becomes eligible to receive a confidential + /// This is how a module peer becomes eligible to receive a confidential /// message: the artifact was hashed against `modules.toml` before `dlopen`, /// and nothing reached across a transport to establish it. #[cfg(feature = "modules")] @@ -292,6 +292,7 @@ impl Broker { &destination, message.header.expected_responder.as_ref(), confidential_call, + &message.header, ); #[cfg(feature = "modules")] let target = target.map_err(|error| { diff --git a/crates/tinybus/src/connection.rs b/crates/tinybus/src/connection.rs index a2bf84f6..d4d4af7d 100644 --- a/crates/tinybus/src/connection.rs +++ b/crates/tinybus/src/connection.rs @@ -1575,7 +1575,7 @@ async fn dispatch(inner: &Inner, header: &Header, body: Value, brokered: bool) - return Err(Error::protocol("method call is missing an address")); }; let context = if brokered { - crate::CallContext::brokered(header.sender.as_ref()) + crate::CallContext::brokered(header.sender.as_ref(), header.confidential) } else { crate::CallContext::default() }; diff --git a/crates/tinybus/src/message/mod.rs b/crates/tinybus/src/message/mod.rs index 1e3ec670..6e74d0c8 100644 --- a/crates/tinybus/src/message/mod.rs +++ b/crates/tinybus/src/message/mod.rs @@ -84,8 +84,8 @@ pub struct Header { /// For [`MessageKind::Error`]: the stable dotted error name. #[serde(default, skip_serializing_if = "Option::is_none")] pub error_name: Option, - /// This body is a secret: deliver it to the attested destination or to - /// nobody. + /// This body is a secret: deliver it to the attested destination, an exact + /// active SDK-created local host callback capability, or nobody. /// /// Set by the sender and *not* overwritten on ingress, unlike `sender`. The /// asymmetry is deliberate and safe in this direction: the flag only ever diff --git a/crates/tinybus/src/module/host.rs b/crates/tinybus/src/module/host.rs index 341ef6ea..eb9c9cd6 100644 --- a/crates/tinybus/src/module/host.rs +++ b/crates/tinybus/src/module/host.rs @@ -96,6 +96,40 @@ struct LoadedModule { transition_from: Option, descriptor_info: Option, lifecycle_busy: bool, + callback_config: Option<[u8; 32]>, +} + +/// Fingerprints only bytes produced by serde_json::to_vec, never arbitrary JSON. +/// Sorted object keys give stable serialization. Float signed zero is the sole +/// serialization difference between equal Values: normalize its numeric token +/// while hashing borrowed segments, without another plaintext allocation. +fn callback_config_fingerprint(bytes: &[u8]) -> [u8; 32] { + use sha2::{Digest, Sha256}; + let mut hash = Sha256::new(); + let mut quoted = false; + let mut index = 0; + let mut segment = 0; + while index < bytes.len() { + if quoted { + match bytes[index] { + b'\\' => index += 1, + b'"' => quoted = false, + _ => {} + } + } else if bytes[index] == b'"' { + quoted = true; + } else if bytes[index..].starts_with(b"-0.0") + && (index == 0 || matches!(bytes[index - 1], b'[' | b',' | b':')) + && (index + 4 == bytes.len() || matches!(bytes[index + 4], b']' | b',' | b'}')) + { + hash.update(&bytes[segment..index]); + segment = index + 1; + index += 3; + } + index += 1; + } + hash.update(&bytes[segment..]); + hash.finalize().into() } enum PendingModule { @@ -1110,6 +1144,12 @@ impl ModuleHost { { Ok(change) => change, Err(_) => { + self.inner + .broker + .router + .lock() + .expect("router lock") + .detach_unique(&unique); let _ = transport.stop_sync(Duration::from_millis(0)); let error = Error::module_refused(path, "module bus name is already owned"); self.record_manifest_rejection(&error, manifest, RefusalClass::Unresolved); @@ -1160,6 +1200,9 @@ impl ModuleHost { tokio::spawn(async move { if lazy_init { ready_transport.wait_initializing().await; + if !ready_transport.init_started() || ready_transport.is_faulted() { + return; + } broker .announce_module_state(serde_json::json!([ module_name.clone(), @@ -1204,6 +1247,7 @@ impl ModuleHost { transition_from: None, descriptor_info, lifecycle_busy: false, + callback_config: None, }); Ok(admitted) } @@ -1450,6 +1494,11 @@ impl ModuleControl for ModuleHostInner { .to_string(), }); } + self.broker + .router + .lock() + .expect("router lock") + .revoke_callbacks_for_peer(&module.unique_name); module.transition_from = Some(old); module.lifecycle_busy = true; module.transport.clone() @@ -1474,6 +1523,12 @@ impl ModuleControl for ModuleHostInner { } async fn reinitialize(&self, name: &str, config: serde_json::Value) -> Result { + let bytes = crate::Secret::new( + serde_json::to_vec(&config) + .map_err(|_| Error::failed("module configuration is invalid"))?, + ); + drop(config); + let fingerprint = callback_config_fingerprint(bytes.expose_secret()); let transport = { let mut loaded = self.loaded.lock().expect("module list lock"); let module = loaded @@ -1481,6 +1536,13 @@ impl ModuleControl for ModuleHostInner { .find(|module| module.info.name == name) .ok_or_else(|| Error::failed("module is not loaded"))?; let state = module.snapshot().state; + if module + .callback_config + .as_ref() + .is_some_and(|initial| initial != &fingerprint) + { + return Err(Error::failed("scoped callback configuration is immutable")); + } if module.lifecycle_busy { return Err(Error::failed( "module lifecycle operation is already running", @@ -1497,7 +1559,7 @@ impl ModuleControl for ModuleHostInner { module.transport.clone() }; let _lifecycle_busy = LifecycleBusyGuard::for_reinitialize(self, name); - let reinitialized = transport.reinitialize(config).await; + let reinitialized = transport.reinitialize(bytes).await; let mut loaded = self.loaded.lock().expect("module list lock"); let module = loaded .iter_mut() @@ -2674,6 +2736,9 @@ mod tests; #[path = "host_admission_tests.rs"] mod admission_tests; +#[path = "host_callbacks.rs"] +pub(crate) mod callbacks; + #[cfg(test)] #[path = "host_required_tests.rs"] mod required_tests; diff --git a/crates/tinybus/src/module/host_callbacks.rs b/crates/tinybus/src/module/host_callbacks.rs new file mode 100644 index 00000000..1caff6db --- /dev/null +++ b/crates/tinybus/src/module/host_callbacks.rs @@ -0,0 +1,716 @@ +//! Local, exact-instance confidential callbacks; never artifact attestation. + +use super::*; +use crate::message::Header; +use crate::transport::memory::MemoryTransport; +use crate::{CallContext, Connection, Interface, InterfaceName, MemberName, ObjectPath}; +use async_trait::async_trait; +use tokio::sync::Notify; + +/// An exact callback interface and its finite member allowlist. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct HostCallbackEndpoint { + /// Fixed object path; no subtree matching. + pub path: ObjectPath, + /// Fixed interface; no alias matching. + pub interface: InterfaceName, + /// Permitted methods, between one and 64 distinct names. + pub members: Vec, +} + +struct Lifetime { + active: bool, + used: bool, + revoked: bool, + module: Option, + running: usize, + served: Vec<(ObjectPath, InterfaceName)>, + registering: Vec<(ObjectPath, InterfaceName)>, +} + +pub(crate) struct CallbackState { + pub(crate) destination: BusName, + endpoints: Vec, + module_transport: Mutex>>, + lifetime: Mutex, + drained: Notify, +} + +impl CallbackState { + fn module_live(&self) -> bool { + self.module_transport + .lock() + .expect("callback module lock") + .as_ref() + .is_none_or(|module| module.upgrade().is_some_and(|module| !module.is_faulted())) + } + + pub(crate) fn module_peer(&self) -> Option { + self.lifetime + .lock() + .expect("callback lifetime lock") + .module + .clone() + } + + pub(crate) fn includes_peer(&self, peer: &BusName) -> bool { + &self.destination == peer || self.module_peer().as_ref() == Some(peer) + } + + pub(crate) fn revoke(&self) { + let mut state = self.lifetime.lock().expect("callback lifetime lock"); + state.active = false; + state.revoked = true; + } + + pub(crate) fn matches(&self, header: &Header) -> bool { + let state = self.lifetime.lock().expect("callback lifetime lock"); + state.active + && self.module_live() + && header.confidential + && header.sender.as_ref() == state.module.as_ref() + && state.module.is_some() + && header.destination.as_ref() == Some(&self.destination) + && self.endpoints.iter().any(|endpoint| { + header.path.as_ref() == Some(&endpoint.path) + && header.interface.as_ref() == Some(&endpoint.interface) + && header + .member + .as_ref() + .is_some_and(|member| endpoint.members.contains(member)) + }) + } +} + +/// Read-only liveness and authenticated module identity for a callback worker. +/// +/// Check again after awaits and before publishing provider results. Revocation +/// prevents new work but cannot undo a provider write already in progress. +#[derive(Clone)] +pub struct HostCallbackGuard(Arc); + +impl HostCallbackGuard { + /// The exact loaded module peer, installed before initialization can call back. + pub fn module_peer(&self) -> Option { + self.0.module_peer() + } + + /// Whether this binding remains active for this authenticated sender. + pub fn authorizes(&self, context: &CallContext) -> bool { + let state = self.0.lifetime.lock().expect("callback lifetime lock"); + state.active + && self.0.module_live() + && context.is_confidential() + && state.module.is_some() + && context.authenticated_sender() == state.module.as_ref() + } + + /// Number of callbacks whose user futures have not yet completed or dropped. + /// A caller timeout does not remove an executing callback from this count. + pub fn running(&self) -> usize { + self.0 + .lifetime + .lock() + .expect("callback lifetime lock") + .running + } + + /// Wait for currently running user futures to complete or drop. + /// Revoke first to prevent new admission. This proves no provider activity + /// beyond the lifetime of the user future; detached provider tasks need + /// their own completion fence. + pub async fn drained(&self) { + loop { + let changed = self.0.drained.notified(); + if self.running() == 0 { + return; + } + changed.await; + } + } +} + +/// SDK-created local callback recipient, scoped once to one verified module. +/// +/// This is intentionally not an attestation and cannot enroll an existing +/// connection, custom transport, well-known alias, or remote recipient. Drop +/// revokes immediately; explicit close also closes the owned local connection. +pub struct HostCallbackBinding { + owner: std::sync::Weak, + broker: Broker, + connection: Connection, + state: Arc, + transport: Arc, +} + +/// Outgoing operations on the binding's private connection. +/// +/// This handle deliberately has no connection accessor, dereference, +/// conversion, or service registration surface. Clones retain only this API. +#[derive(Clone, Debug)] +pub struct HostCallbackProxy(crate::Proxy); + +impl HostCallbackProxy { + /// Override the deadline for outgoing calls. + pub fn with_timeout(mut self, timeout: Duration) -> Self { + self.0 = self.0.with_timeout(timeout); + self + } + /// The destination addressed by this handle. + pub fn destination(&self) -> &BusName { + self.0.destination() + } + /// The addressed object path. + pub fn path(&self) -> &ObjectPath { + self.0.path() + } + /// The addressed interface. + pub fn interface(&self) -> &InterfaceName { + self.0.interface() + } + /// Call a method and deserialize its ordinary reply. + pub async fn call( + &self, + member: &str, + args: impl serde::Serialize, + ) -> Result { + self.0.call(member, args).await + } + /// Call a method receiving a bounded streaming reply. + pub async fn call_streaming( + &self, + member: &str, + args: impl serde::Serialize, + ) -> Result { + self.0.call_streaming(member, args).await + } + /// Call an attested recipient with a confidential body. + pub async fn call_confidential( + &self, + member: &str, + args: impl serde::Serialize, + ) -> Result { + self.0.call_confidential(member, args).await + } + /// Call a broker module-control method with a redacted body. + pub async fn call_sensitive( + &self, + member: &str, + args: impl serde::Serialize, + ) -> Result { + self.0.call_sensitive(member, args).await + } + /// Read the broker's artifact attestation for the destination. + pub async fn attestation(&self) -> Result> { + self.0.attestation().await + } + /// Whether the destination currently has an owner. + pub async fn is_available(&self) -> Result { + self.0.is_available().await + } + /// Subscribe to signals for the addressed object and interface. + pub async fn receive_signal( + &self, + member: &str, + ) -> Result> { + self.0.receive_signal(member).await + } +} + +impl HostCallbackBinding { + /// The exact SDK-created host destination for module configuration. + pub fn unique_name(&self) -> &BusName { + &self.state.destination + } + /// Construct an outgoing proxy using this private host connection. + /// + /// The outgoing handle cannot expose the owned callback connection: + /// ```compile_fail + /// # async fn example(binding: &tinybus::module::HostCallbackBinding) -> tinybus::Result<()> { + /// let outgoing = binding.proxy("org.example.Module", "/module", "org.example.Module")?; + /// let _: &tinybus::Connection = outgoing.connection(); + /// # Ok(()) + /// # } + /// ``` + /// The proxy exposes no raw connection or service registration escape. + pub fn proxy( + &self, + destination: &str, + path: &str, + interface: &str, + ) -> Result { + self.connection + .proxy(destination, path, interface) + .map(HostCallbackProxy) + } + /// Read-only capability state for domain authorization and completion accounting. + pub fn guard(&self) -> HostCallbackGuard { + HostCallbackGuard(self.state.clone()) + } + /// Immediately and idempotently prevent new route and user-dispatch admission. + /// Running user futures remain accounted for until actual completion/drop. + pub fn revoke(&self) { + self.broker + .router + .lock() + .expect("router lock") + .revoke_host_callbacks(&self.state); + } + /// Revoke and close the local connection. This does not cancel running user code. + pub async fn close(&self) -> Result<()> { + self.revoke(); + self.connection.close().await + } + /// Serve a declared endpoint with an unbypassable authenticated live guard. + /// Every declared member must be implemented; duplicate registration fails. + pub async fn serve_at(&self, path: ObjectPath, interface: impl Interface) -> Result<()> { + let name = interface.name(); + let endpoint = self + .state + .endpoints + .iter() + .find(|endpoint| endpoint.path == path && endpoint.interface == name) + .ok_or_else(|| Error::failed("callback endpoint is not declared"))?; + let provided = interface.members(); + if !endpoint + .members + .iter() + .all(|member| provided.contains(member)) + { + return Err(Error::failed("callback endpoint members are missing")); + } + { + let mut state = self.state.lifetime.lock().expect("callback lifetime lock"); + if state.used + || state.revoked + || state.served.contains(&(path.clone(), name.clone())) + || state.registering.contains(&(path.clone(), name.clone())) + { + return Err(Error::failed( + "callback registration is immutable or duplicated", + )); + } + state.registering.push((path.clone(), name.clone())); + } + let result = self + .connection + .serve_at( + path.clone(), + GuardedInterface { + interface, + endpoint: endpoint.clone(), + state: self.state.clone(), + }, + ) + .await; + { + let mut state = self.state.lifetime.lock().expect("callback lifetime lock"); + state + .registering + .retain(|entry| entry != &(path.clone(), name.clone())); + if result.is_ok() { + state.served.push((path, name)); + } + } + result + } +} + +impl Drop for HostCallbackBinding { + fn drop(&mut self) { + self.revoke(); + // Drop cannot await. Closing the sender synchronously detaches the + // owned peer without spawning a task that might never be polled. + self.transport.close_outbound(); + self.broker + .router + .lock() + .expect("router lock") + .detach_unique(&self.state.destination); + } +} + +struct GuardedInterface { + interface: I, + endpoint: HostCallbackEndpoint, + state: Arc, +} + +struct Running(Arc); +impl Drop for Running { + fn drop(&mut self) { + let mut state = self.0.lifetime.lock().expect("callback lifetime lock"); + state.running -= 1; + if state.running == 0 { + self.0.drained.notify_waiters(); + } + } +} + +#[async_trait] +impl Interface for GuardedInterface { + fn name(&self) -> InterfaceName { + self.endpoint.interface.clone() + } + fn members(&self) -> Vec { + self.endpoint.members.clone() + } + async fn call(&self, _: &MemberName, _: serde_json::Value) -> Result { + Err(Error::failed("authenticated scoped callback required")) + } + async fn call_with_context( + &self, + member: &MemberName, + args: serde_json::Value, + context: &CallContext, + ) -> Result { + // Serialized with revoke: queued deliveries cannot bypass the live + // check by carrying an earlier route admission through an await. + let _running = { + let mut state = self.state.lifetime.lock().expect("callback lifetime lock"); + if !self.state.module_live() + || !context.is_confidential() + || !state.active + || state.module.is_none() + || context.authenticated_sender() != state.module.as_ref() + || !self.endpoint.members.contains(member) + { + return Err(Error::failed("scoped callback authority is unavailable")); + } + state.running += 1; + Running(self.state.clone()) + }; + self.interface + .call_with_context(member, args, context) + .await + } +} + +impl ModuleHost { + /// Create a private local recipient for at most 32 exact callback interfaces. + /// No authority exists until a verified module is loaded with this binding. + pub async fn prepare_host_callbacks( + &self, + endpoints: Vec, + ) -> Result { + if endpoints.is_empty() + || endpoints.len() > 32 + || endpoints.iter().any(|endpoint| { + endpoint.members.is_empty() + || endpoint.members.len() > 64 + || endpoint + .members + .iter() + .enumerate() + .any(|(i, member)| endpoint.members[..i].contains(member)) + }) + || endpoints.iter().enumerate().any(|(i, endpoint)| { + endpoints[..i].iter().any(|previous| { + previous.path == endpoint.path && previous.interface == endpoint.interface + }) + }) + { + return Err(Error::failed("callback endpoint declaration is invalid")); + } + let (host, peer) = MemoryTransport::pair(); + let unique = self.inner.broker.attach(Arc::new(peer)); + let transport = Arc::new(host); + let mut rollback = PeerRollback { + broker: self.inner.broker.clone(), + peer: unique, + transport: transport.clone(), + committed: false, + }; + let connection = Connection::attach(transport.clone()); + connection.handshake().await?; + let destination = connection + .unique_name() + .ok_or_else(|| Error::failed("callback connection has no unique name"))?; + rollback.committed = true; + Ok(HostCallbackBinding { + owner: Arc::downgrade(&self.inner), + broker: self.inner.broker.clone(), + connection, + transport, + state: Arc::new(CallbackState { + destination, + endpoints, + module_transport: Mutex::new(None), + lifetime: Mutex::new(Lifetime { + active: false, + used: false, + revoked: false, + module: None, + running: 0, + served: Vec::new(), + registering: Vec::new(), + }), + drained: Notify::new(), + }), + }) + } + + /// Load a hash-verified file and bind its exact peer before any initialization. + /// + /// All endpoints must already be served. The binding is single-use even on + /// failure. Configuration is frozen for reinitialization. Generic load APIs + /// grant no callback authority. The caller supplies the binding's unique + /// destination in its domain-specific configuration; it cannot change scope. + pub async fn load_file_with_host_callbacks( + &self, + path: impl AsRef, + config: serde_json::Value, + binding: &HostCallbackBinding, + ) -> Result { + if !Arc::ptr_eq(&self.inner.broker.router, &binding.broker.router) + || !binding + .owner + .upgrade() + .is_some_and(|owner| Arc::ptr_eq(&owner, &self.inner)) + { + return Err(Error::failed( + "callback binding belongs to another host owner", + )); + } + { + let mut state = binding + .state + .lifetime + .lock() + .expect("callback lifetime lock"); + if state.used || state.revoked || state.served.len() != binding.state.endpoints.len() { + return Err(Error::failed( + "callback binding is already used or endpoints are missing", + )); + } + state.used = true; + } + let mut rollback = LoadRollback { + binding, + host: self, + transport: None, + committed: false, + }; + let path = path.as_ref(); + if let Some(parent) = path.parent() { + check_directory(parent)?; + } + check_file(path)?; + let hash = allowlisted_hash(path, std::fs::File::open(path)?)?.ok_or_else(|| { + Error::module_refused(path, "scoped callbacks require a verified artifact") + })?; + let artifact = loader::load(path, self.inner.strict.load(Ordering::Acquire))?; + self.ensure_dependencies(&artifact.manifest, path)?; + if allowlisted_hash(path, std::fs::File::open(path)?)?.as_ref() != Some(&hash) { + return Err(Error::module_refused( + path, + "verified artifact changed during loading", + )); + } + let transport = { + let _admission = self.inner.admission.lock().expect("module admission lock"); + let admitted = self.validate(path, &artifact.descriptor, &artifact.manifest)?; + if self + .inner + .loaded + .lock() + .expect("module list lock") + .iter() + .any(|module| module.info.name == admitted.name) + { + return Err(Error::module_refused(path, "module name is already loaded")); + } + // Reservation happens inside attach_transport. Initialization is + // deferred even for an eager module until routes and identity exist. + let bytes = serde_json::to_vec(&config) + .map_err(|_| Error::failed("module configuration is invalid"))?; + let fingerprint = callback_config_fingerprint(&bytes); + drop(config); + let (transport, host) = ModuleTransport::new(admitted.name.clone(), bytes); + transport.pause_initialization(); + transport.defer_initialize(artifact.init, host); + let name = admitted.name.clone(); + rollback.transport = Some(transport.clone()); + self.attach_transport( + path, + admitted, + artifact.manifest, + transport.clone(), + Activation { + lazy_init: true, + lazy_load: false, + descriptor_info: None, + pinned_sha256: Some(hash), + }, + )?; + rollback.transport = Some(transport.clone()); + let mut loaded = self.inner.loaded.lock().expect("module list lock"); + let module = loaded + .iter_mut() + .find(|module| module.info.name == name) + .expect("module was attached"); + { + let mut state = binding + .state + .lifetime + .lock() + .expect("callback lifetime lock"); + if state.revoked || state.module.is_some() { + return Err(Error::failed("callback binding was concurrently used")); + } + *binding + .state + .module_transport + .lock() + .expect("callback module lock") = Some(Arc::downgrade(&transport)); + state.module = Some(module.unique_name.clone()); + state.active = true; + } + module.callback_config = Some(fingerprint); + self.inner + .broker + .router + .lock() + .expect("router lock") + .add_host_callbacks(binding.state.clone())?; + transport.release_initialization(true); + transport + }; + let initialized = transport.ensure_initialized().await; + let ready = if initialized.is_ok() { + tokio::time::timeout(Duration::from_secs(5), transport.wait_ready()) + .await + .is_ok() + && transport.is_ready() + } else { + false + }; + if !ready { + binding.revoke(); + let _ = transport.stop_sync(Duration::ZERO); + let _ = transport.close().await; + self.inner + .loaded + .lock() + .expect("module list lock") + .retain(|module| !Arc::ptr_eq(&module.transport, &transport)); + return Err(Error::module_refused(path, "module initialization failed")); + } + let loaded = self.inner.loaded.lock().expect("module list lock"); + let info = loaded + .iter() + .find(|module| Arc::ptr_eq(&module.transport, &transport)) + .ok_or_else(|| { + Error::module_refused(path, "module was removed before load completion") + })? + .snapshot(); + let lifetime = binding + .state + .lifetime + .lock() + .expect("callback lifetime lock"); + // Snapshot first, then validate liveness at the commit point. A fault + // racing this point may be a subsequent lifecycle event, but must never + // turn the returned successful snapshot into Faulted or Stopped. + if lifetime.revoked + || !lifetime.active + || !transport.is_ready() + || !matches!(info.state, ModuleState::Ready) + { + return Err(Error::module_refused( + path, + "module became unavailable before load completion", + )); + } + rollback.committed = true; + Ok(info) + } +} + +struct LoadRollback<'a> { + binding: &'a HostCallbackBinding, + host: &'a ModuleHost, + transport: Option>, + committed: bool, +} +impl Drop for LoadRollback<'_> { + fn drop(&mut self) { + if self.committed { + return; + } + self.binding.revoke(); + self.binding.transport.close_outbound(); + self.binding + .broker + .router + .lock() + .expect("router lock") + .detach_unique(&self.binding.state.destination); + if let Some(transport) = &self.transport { + transport.release_initialization(false); + let peer = self.binding.state.module_peer().or_else(|| { + self.host + .inner + .loaded + .lock() + .expect("module list lock") + .iter() + .find(|module| Arc::ptr_eq(&module.transport, transport)) + .map(|module| module.unique_name.clone()) + }); + if let Some(peer) = peer { + self.host + .inner + .broker + .router + .lock() + .expect("router lock") + .detach_unique(&peer); + } + // A timed-out initializer may still run. Stop is best effort; + // revocation/detach removes authority without claiming cessation. + let _ = transport.stop_sync(Duration::ZERO); + self.host + .inner + .loaded + .lock() + .expect("module list lock") + .retain(|module| !Arc::ptr_eq(&module.transport, transport)); + } + } +} + +impl Drop for ModuleHostInner { + fn drop(&mut self) { + for module in self.loaded.get_mut().expect("module list lock").iter() { + self.broker + .router + .lock() + .expect("router lock") + .revoke_callbacks_for_peer(&module.unique_name); + } + } +} + +struct PeerRollback { + broker: Broker, + peer: BusName, + transport: Arc, + committed: bool, +} +impl Drop for PeerRollback { + fn drop(&mut self) { + if !self.committed { + self.transport.close_outbound(); + self.broker + .router + .lock() + .expect("router lock") + .detach_unique(&self.peer); + } + } +} + +#[cfg(test)] +#[path = "host_callbacks_tests.rs"] +mod tests; diff --git a/crates/tinybus/src/module/host_callbacks_tests.rs b/crates/tinybus/src/module/host_callbacks_tests.rs new file mode 100644 index 00000000..4ddd7b2f --- /dev/null +++ b/crates/tinybus/src/module/host_callbacks_tests.rs @@ -0,0 +1,558 @@ +use super::*; +use crate::message::Message; +use crate::router::Router; +use std::sync::atomic::AtomicUsize; + +fn endpoint() -> HostCallbackEndpoint { + HostCallbackEndpoint { + path: "/callback".parse().unwrap(), + interface: "org.example.Callback".parse().unwrap(), + members: vec!["Observe".parse().unwrap()], + } +} +fn state(source: BusName, destination: BusName) -> Arc { + Arc::new(CallbackState { + destination, + endpoints: vec![endpoint()], + module_transport: Mutex::new(None), + lifetime: Mutex::new(Lifetime { + active: true, + used: true, + revoked: false, + module: Some(source), + running: 0, + served: vec![], + registering: vec![], + }), + drained: Notify::new(), + }) +} +fn call(state: &CallbackState) -> Message { + let endpoint = endpoint(); + let mut message = Message::method_call( + state.destination.clone(), + endpoint.path, + endpoint.interface, + endpoint.members[0].clone(), + serde_json::json!([]), + ); + message.header.sender = state.module_peer(); + message.header.confidential = true; + message +} +fn kernel_fixture() -> (Router, Arc, u64, u64) { + // Private router behavior tests only; actual artifact admission is proved by + // host_tests with the hash-verified cdylib, never by this table fixture. + let mut router = Router::default(); + let (source_id, source) = router.attach(tokio::sync::mpsc::channel(8).0); + let (destination_id, destination) = router.attach(tokio::sync::mpsc::channel(8).0); + router.set_attestation_for_unique( + &source, + crate::attest::Attestation { + name: "org.example.Module".parse().unwrap(), + sha256: "a".repeat(64), + }, + ); + let state = state(source, destination); + router.add_host_callbacks(state.clone()).unwrap(); + (router, state, source_id, destination_id) +} +fn resolves(router: &Router, message: &Message) -> Result> { + router.resolve_call( + message.header.destination.as_ref().unwrap(), + message.header.expected_responder.as_ref(), + true, + &message.header, + ) +} +#[test] +fn scoped_routes_match_every_field_and_never_attest_the_host() { + let (router, state, _, _) = kernel_fixture(); + let message = call(&state); + assert!(resolves(&router, &message).is_ok()); + assert!(router.attestation_of(&state.destination).is_none()); + assert!(router.resolve_attested(&state.destination).is_err()); + for field in 0..5 { + let mut wrong = message.clone(); + match field { + 0 => wrong.header.sender = Some(":1.999".parse().unwrap()), + 1 => wrong.header.path = Some("/other".parse().unwrap()), + 2 => wrong.header.interface = Some("org.example.Other".parse().unwrap()), + 3 => wrong.header.member = Some("Other".parse().unwrap()), + _ => wrong.header.expected_responder = Some(":1.999".parse().unwrap()), + } + assert!(resolves(&router, &wrong).is_err()); + } +} +#[test] +fn a_host_alias_and_a_second_verified_module_inherit_no_authority() { + let (mut router, state, _, destination_id) = kernel_fixture(); + let (_, other) = router.attach(tokio::sync::mpsc::channel(8).0); + router.set_attestation_for_unique( + &other, + crate::attest::Attestation { + name: "org.example.Other".parse().unwrap(), + sha256: "b".repeat(64), + }, + ); + let mut message = call(&state); + message.header.sender = Some(other); + assert!(resolves(&router, &message).is_err()); + let alias: BusName = "org.example.Host".parse().unwrap(); + router.request_name(destination_id, alias.clone()).unwrap(); + message = call(&state); + message.header.destination = Some(alias); + assert!(resolves(&router, &message).is_err()); +} +#[test] +fn detaching_either_peer_revokes_stale_guard_and_replacements() { + for detach_source in [false, true] { + let (mut router, state, source_id, destination_id) = kernel_fixture(); + router.detach(if detach_source { + source_id + } else { + destination_id + }); + assert!(!state.matches(&call(&state).header)); + assert_eq!( + router + .add_host_callbacks(state.clone()) + .unwrap_err() + .wire_name(), + "ai.tinyhumans.tinybus.Error.Failed" + ); + let (_, replacement) = router.attach(tokio::sync::mpsc::channel(8).0); + assert_ne!(Some(replacement), state.module_peer()); + } +} +#[test] +fn unverified_sources_cannot_be_enrolled_and_scopes_are_bounded() { + let mut router = Router::default(); + let (_, source) = router.attach(tokio::sync::mpsc::channel(8).0); + let (_, destination) = router.attach(tokio::sync::mpsc::channel(8).0); + let state = state(source.clone(), destination); + assert!(router.add_host_callbacks(state.clone()).is_err()); + router.set_attestation_for_unique( + &source, + crate::attest::Attestation { + name: "org.example.Module".parse().unwrap(), + sha256: "a".repeat(64), + }, + ); + router.add_host_callbacks(state.clone()).unwrap(); + assert!(router.add_host_callbacks(state).is_err()); + for _ in 1..128 { + let (_, source) = router.attach(tokio::sync::mpsc::channel(8).0); + let (_, destination) = router.attach(tokio::sync::mpsc::channel(8).0); + router.set_attestation_for_unique( + &source, + crate::attest::Attestation { + name: "org.example.Module".parse().unwrap(), + sha256: "a".repeat(64), + }, + ); + router + .add_host_callbacks(super::tests::state(source, destination)) + .unwrap(); + } + let (_, source) = router.attach(tokio::sync::mpsc::channel(8).0); + let (_, destination) = router.attach(tokio::sync::mpsc::channel(8).0); + router.set_attestation_for_unique( + &source, + crate::attest::Attestation { + name: "org.example.Module".parse().unwrap(), + sha256: "a".repeat(64), + }, + ); + let extra = super::tests::state(source, destination); + assert!(router.add_host_callbacks(extra).is_err()); +} + +struct BlockingCallback { + entered: tokio::sync::mpsc::Sender<()>, + release: Arc, + published: Arc, + guard: HostCallbackGuard, +} +#[async_trait] +impl Interface for BlockingCallback { + fn name(&self) -> InterfaceName { + endpoint().interface + } + fn members(&self) -> Vec { + endpoint().members + } + async fn call(&self, _: &MemberName, _: serde_json::Value) -> Result { + Err(Error::failed("unverified")) + } + async fn call_with_context( + &self, + _: &MemberName, + _: serde_json::Value, + context: &CallContext, + ) -> Result { + self.entered.send(()).await.unwrap(); + let _permit = self.release.acquire().await.unwrap(); + if !self.guard.authorizes(context) { + return Err(Error::failed("revoked before publication")); + } + self.published.fetch_add(1, Ordering::SeqCst); + Ok(serde_json::Value::Null) + } +} +#[tokio::test] +async fn revoke_after_route_selection_blocks_dispatch_and_direct_unverified_dispatch() { + let (mut router, state, _, _) = kernel_fixture(); + let message = call(&state); + assert!(resolves(&router, &message).is_ok()); + let (entered, mut receiver) = tokio::sync::mpsc::channel(1); + let guarded = GuardedInterface { + endpoint: endpoint(), + state: state.clone(), + interface: BlockingCallback { + entered, + release: Arc::new(tokio::sync::Semaphore::new(1)), + published: Arc::new(AtomicUsize::new(0)), + guard: HostCallbackGuard(state.clone()), + }, + }; + assert!( + guarded + .call(&endpoint().members[0], serde_json::json!([])) + .await + .is_err() + ); + assert!( + guarded + .call_with_context( + &endpoint().members[0], + serde_json::json!([]), + &CallContext::brokered(state.module_peer().as_ref(), false) + ) + .await + .is_err() + ); + assert!( + guarded + .call_with_context( + &endpoint().members[0], + serde_json::json!([]), + &CallContext::default() + ) + .await + .is_err() + ); + router.revoke_host_callbacks(&state); + assert!( + guarded + .call_with_context( + &endpoint().members[0], + serde_json::json!([]), + &CallContext::brokered(state.module_peer().as_ref(), true) + ) + .await + .is_err() + ); + assert!(receiver.try_recv().is_err()); + assert_eq!(HostCallbackGuard(state).running(), 0); +} +#[tokio::test] +async fn revoke_during_await_preserves_running_charge_until_actual_completion() { + let (mut router, state, _, _) = kernel_fixture(); + let guard = HostCallbackGuard(state.clone()); + let (entered, mut receiver) = tokio::sync::mpsc::channel(1); + let release = Arc::new(tokio::sync::Semaphore::new(0)); + let published = Arc::new(AtomicUsize::new(0)); + let guarded = GuardedInterface { + endpoint: endpoint(), + state: state.clone(), + interface: BlockingCallback { + entered, + release: release.clone(), + published: published.clone(), + guard: guard.clone(), + }, + }; + let context = CallContext::brokered(state.module_peer().as_ref(), true); + let task = tokio::spawn(async move { + guarded + .call_with_context(&endpoint().members[0], serde_json::json!([]), &context) + .await + }); + receiver.recv().await.unwrap(); + assert_eq!(guard.running(), 1); + router.revoke_host_callbacks(&state); + assert_eq!(guard.running(), 1); + let mut drain = Box::pin(guard.drained()); + assert!( + std::future::poll_fn(|cx| std::task::Poll::Ready(drain.as_mut().poll(cx).is_pending())) + .await + ); + release.add_permits(1); + assert!(task.await.unwrap().is_err()); + guard.drained().await; + assert_eq!(guard.running(), 0); + assert_eq!(published.load(Ordering::SeqCst), 0); +} +#[tokio::test] +async fn endpoint_declarations_registration_and_broker_ownership_are_checked() { + let host = ModuleHost::new(Broker::new()); + assert!(host.prepare_host_callbacks(vec![]).await.is_err()); + assert!( + host.prepare_host_callbacks(vec![endpoint(); 33]) + .await + .is_err() + ); + assert!( + host.prepare_host_callbacks(vec![endpoint(), endpoint()]) + .await + .is_err() + ); + let mut duplicate = endpoint(); + duplicate.members.push(duplicate.members[0].clone()); + assert!(host.prepare_host_callbacks(vec![duplicate]).await.is_err()); + let mut excessive = endpoint(); + excessive.members = (0..65).map(|i| format!("M{i}").parse().unwrap()).collect(); + assert!(host.prepare_host_callbacks(vec![excessive]).await.is_err()); + let binding = host.prepare_host_callbacks(vec![endpoint()]).await.unwrap(); + assert!( + host.load_file_with_host_callbacks("unused", serde_json::json!({}), &binding) + .await + .is_err() + ); + let other_owner = ModuleHost::new(host.inner.broker.clone()); + assert!( + other_owner + .load_file_with_host_callbacks("unused", serde_json::json!({}), &binding) + .await + .is_err() + ); + let other = ModuleHost::new(Broker::new()); + assert!( + other + .load_file_with_host_callbacks("unused", serde_json::json!({}), &binding) + .await + .is_err() + ); + binding.revoke(); + binding.revoke(); + assert!( + host.load_file_with_host_callbacks("unused", serde_json::json!({}), &binding) + .await + .is_err() + ); + binding.close().await.unwrap(); +} + +struct EmptyCallback; +#[async_trait] +impl Interface for EmptyCallback { + fn name(&self) -> InterfaceName { + endpoint().interface + } + fn members(&self) -> Vec { + vec![] + } + async fn call(&self, _: &MemberName, _: serde_json::Value) -> Result { + Ok(serde_json::Value::Null) + } +} +#[tokio::test] +async fn registration_is_exact_single_use_and_failed_load_detaches_the_local_peer() { + let broker = Broker::new(); + let host = ModuleHost::new(broker.clone()); + let binding = host.prepare_host_callbacks(vec![endpoint()]).await.unwrap(); + assert!( + binding + .serve_at("/other".parse().unwrap(), EmptyCallback) + .await + .is_err() + ); + assert!( + binding + .serve_at(endpoint().path, EmptyCallback) + .await + .is_err() + ); + let (entered, _) = tokio::sync::mpsc::channel(1); + let callback = || BlockingCallback { + entered: entered.clone(), + release: Arc::new(tokio::sync::Semaphore::new(1)), + published: Arc::new(AtomicUsize::new(0)), + guard: binding.guard(), + }; + binding.serve_at(endpoint().path, callback()).await.unwrap(); + assert!(binding.serve_at(endpoint().path, callback()).await.is_err()); + assert_eq!( + binding.unique_name(), + &broker + .router + .lock() + .unwrap() + .owner_of(binding.unique_name()) + .unwrap() + ); + assert!( + host.load_file_with_host_callbacks("missing-fixture", serde_json::json!({}), &binding) + .await + .is_err() + ); + assert!( + broker + .router + .lock() + .unwrap() + .owner_of(binding.unique_name()) + .is_none() + ); + assert!(binding.serve_at(endpoint().path, callback()).await.is_err()); + assert!( + host.load_file_with_host_callbacks("missing-fixture", serde_json::json!({}), &binding) + .await + .is_err() + ); + let peer = binding.unique_name().clone(); + drop(binding); + assert!(broker.router.lock().unwrap().owner_of(&peer).is_none()); +} + +#[tokio::test] +async fn a_module_fault_revokes_dispatch_authority_before_peer_detach() { + let (router, state, _, _) = kernel_fixture(); + let (transport, host) = ModuleTransport::new("fault-fixture".into(), Vec::new()); + *state.module_transport.lock().unwrap() = Some(Arc::downgrade(&transport)); + assert!(resolves(&router, &call(&state)).is_ok()); + unsafe { + (host.fault)(host.host_ctx, std::ptr::null(), 0); + } + assert!(!state.matches(&call(&state).header)); + assert!( + !HostCallbackGuard(state.clone()) + .authorizes(&CallContext::brokered(state.module_peer().as_ref(), true)) + ); + assert!(resolves(&router, &call(&state)).is_err()); +} + +#[tokio::test] +async fn restricted_outgoing_handle_preserves_typed_calls_and_metadata() { + let broker = Broker::new(); + let host = ModuleHost::new(broker.clone()); + let binding = host.prepare_host_callbacks(vec![endpoint()]).await.unwrap(); + let outgoing = binding + .proxy(crate::BUS_NAME, crate::BUS_PATH, crate::BUS_NAME) + .unwrap() + .with_timeout(Duration::from_secs(1)); + assert_eq!(outgoing.destination().as_str(), crate::BUS_NAME); + assert_eq!(outgoing.path().as_str(), crate::BUS_PATH); + assert_eq!(outgoing.interface().as_str(), crate::BUS_NAME); + let names: Vec = outgoing.clone().call("ListNames", ()).await.unwrap(); + assert!(names.contains(binding.unique_name())); + let streamed: Vec = outgoing.call_streaming("ListNames", ()).await.unwrap(); + assert_eq!(names, streamed); + assert!( + outgoing + .call_confidential::("ListNames", ()) + .await + .is_err() + ); + assert!( + outgoing + .call_sensitive::("ListNames", ()) + .await + .is_err() + ); + assert!(outgoing.attestation().await.unwrap().is_none()); + assert!(!outgoing.is_available().await.unwrap()); + let _signals = outgoing.receive_signal("NameOwnerChanged").await.unwrap(); + assert!(format!("{outgoing:?}").contains("HostCallbackProxy")); + binding.close().await.unwrap(); +} + +#[tokio::test] +async fn rollback_after_attach_releases_the_actual_initialization_observer() { + let broker = Broker::new(); + let host = ModuleHost::new(broker); + let binding = host.prepare_host_callbacks(vec![endpoint()]).await.unwrap(); + let manifest = super::super::tests::manifest(); + let path = Path::new("observer-fixture.so"); + let mut admitted = host + .validate(path, &TbAbiDescriptor::current("clock", "0.1.0"), &manifest) + .unwrap(); + admitted.state = ModuleState::Resolved; + let (transport, _) = ModuleTransport::new(admitted.name.clone(), vec![]); + transport.pause_initialization(); + let weak = Arc::downgrade(&transport); + host.attach_transport( + path, + admitted, + manifest, + transport.clone(), + Activation { + lazy_init: true, + lazy_load: false, + descriptor_info: None, + pinned_sha256: None, + }, + ) + .unwrap(); + tokio::task::yield_now().await; + drop(LoadRollback { + binding: &binding, + host: &host, + transport: Some(transport.clone()), + committed: false, + }); + assert!(!transport.init_started()); + assert!(host.list().is_empty()); + drop(transport); + tokio::time::timeout(Duration::from_millis(100), async { + while weak.upgrade().is_some() { + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); +} + +#[test] +fn configuration_fingerprints_preserve_json_equality_without_normalizing_strings() { + fn fingerprint(value: &serde_json::Value) -> [u8; 32] { + let bytes = crate::Secret::new(serde_json::to_vec(value).unwrap()); + callback_config_fingerprint(bytes.expose_secret()) + } + let negative = serde_json::json!({ + "z": [-0.0, {"number": -0.0, "text": "\"-0.0\"\\-0.0"}], + "credential": "private-test-secret", + }); + // Construct the same map in reverse insertion order. + let mut positive = serde_json::Map::new(); + positive.insert( + "credential".into(), + serde_json::json!("private-test-secret"), + ); + positive.insert( + "z".into(), + serde_json::json!([ + 0.0, {"number": 0.0, "text": "\"-0.0\"\\-0.0"} + ]), + ); + let positive = serde_json::Value::Object(positive); + assert_eq!(negative, positive); + assert_eq!(fingerprint(&negative), fingerprint(&positive)); + for differing in [ + serde_json::json!({"z": [0, {"number": 0.0, "text": "\"-0.0\"\\-0.0"}], "credential": "private-test-secret"}), + serde_json::json!({"z": [0.0, {"number": 0.0, "text": "\"0.0\"\\-0.0"}], "credential": "private-test-secret"}), + serde_json::json!({"z": [0.0, {"number": 0.0, "text": "\"-0.0\"\\-0.0"}], "credential": "different-private-test-secret"}), + ] { + assert_ne!(negative, differing); + assert_ne!(fingerprint(&negative), fingerprint(&differing)); + } + assert_eq!( + fingerprint(&serde_json::json!(-0.0)), + fingerprint(&serde_json::json!(0.0)) + ); + assert_ne!( + fingerprint(&serde_json::json!(-0.01)), + fingerprint(&serde_json::json!(0.01)) + ); +} diff --git a/crates/tinybus/src/module/host_tests.rs b/crates/tinybus/src/module/host_tests.rs index 485e31de..2327f787 100644 --- a/crates/tinybus/src/module/host_tests.rs +++ b/crates/tinybus/src/module/host_tests.rs @@ -2182,24 +2182,12 @@ async fn one_refused_module_does_not_stop_the_others_in_the_directory_from_loadi ); } -/// Copy `artifact` into a fresh directory beside a `modules.toml` listing -/// `hash` for it, so a load can be driven against a real allowlist. +/// Copy `artifact` into a fresh, private directory with its hash allowlist. /// -/// Unix-only, and the reason is the loader's own admission check rather than -/// anything about the code under test. On Windows that check trusts exactly -/// three SIDs on a module directory: its owner, `LocalSystem`, and -/// `BUILTIN\Administrators`. A CI runner's account is an administrator, so a -/// directory a *test* creates is owned by `BUILTIN\Administrators` while the -/// ACE it inherits names the user SID — neither the owner nor well-known -/// trusted — and the load is refused. That is why the Windows workflow calls -/// `SetOwner` on the directories it provisions; a test cannot do the same -/// without Win32 calls of its own. -/// -/// Gating here costs little: what these two tests exercise is the hash -/// comparison and the attestation record it produces, which is identical on -/// every platform. The Windows-specific directory policy is covered by the -/// existing loader tests that run against the CI-provisioned directories. -#[cfg(unix)] +/// Each native fixture needs a separate mapping: the SDK installs a global +/// tracing subscriber inside the library, so initializing the same DLL path +/// twice shares state even across independent hosts. Windows staging uses the +/// release-cache ACL repair so the real admission gate remains in force. fn staged_module(artifact: &Path, hash: &str) -> (tempfile::TempDir, PathBuf) { // Staged beside the artifact rather than in `/tmp`: the loader refuses a // directory another user could write to, and `/tmp` is exactly that. The @@ -2209,6 +2197,8 @@ fn staged_module(artifact: &Path, hash: &str) -> (tempfile::TempDir, PathBuf) { // route around. let root = artifact.parent().expect("artifact has a parent directory"); let dir = tempfile::tempdir_in(root).unwrap(); + #[cfg(windows)] + crate::module::cache::secure_release_cache(dir.path(), dir.path()); let file_name = artifact.file_name().unwrap(); let staged = dir.path().join(file_name); std::fs::copy(artifact, &staged).unwrap(); @@ -2410,7 +2400,6 @@ async fn an_eager_cdylib_authenticates_after_its_worker_runs_before_admission() async fn authenticates_native_context(variable: &str) { let artifact = PathBuf::from(std::env::var_os(variable).expect(variable)); let hash = crate::module::hash::file_hex(std::fs::File::open(&artifact).unwrap()).unwrap(); - #[cfg(unix)] let (_dir, artifact) = staged_module(&artifact, &hash); let bus = MemoryBus::new(); let broker = Broker::new(); @@ -2531,3 +2520,367 @@ async fn fixed_unique_module_calls_are_refused_in_terminal_states_before_detach( } broker_task.abort(); } + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +#[ignore = "requires TINYBUS_TEST_CONTEXT_MODULE to point at the built cdylib"] +async fn a_verified_cdylib_can_deliver_a_confidential_host_callback() { + let artifact = PathBuf::from(std::env::var_os("TINYBUS_TEST_CONTEXT_MODULE").unwrap()); + let hash = crate::module::hash::file_hex(std::fs::File::open(&artifact).unwrap()).unwrap(); + let (_dir, artifact) = staged_module(&artifact, &hash); + let bus = MemoryBus::new(); + let broker = Broker::new(); + let task = broker.spawn(bus.clone()); + let modules = ModuleHost::new(broker); + let binding = modules + .prepare_host_callbacks(vec![crate::module::HostCallbackEndpoint { + path: "/callback".parse().unwrap(), + interface: "org.example.Callback".parse().unwrap(), + members: vec!["Observe".parse().unwrap()], + }]) + .await + .unwrap(); + let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + binding + .serve_at( + "/callback".parse().unwrap(), + CountingNativeCallback(calls.clone()), + ) + .await + .unwrap(); + modules + .load_file_with_host_callbacks(artifact, serde_json::json!({}), &binding) + .await + .unwrap(); + let module = binding.guard().module_peer().unwrap(); + let result = binding + .proxy(module.as_str(), "/context", "org.example.Context") + .unwrap() + .call_confidential::<(crate::BusName, crate::BusName)>("ConfidentialCallback", ()) + .await; + assert_eq!( + calls.load(Ordering::SeqCst), + 1, + "confidential callback required" + ); + assert_eq!( + result.unwrap(), + (binding.unique_name().clone(), module.clone()) + ); + let ordinary = binding + .proxy(module.as_str(), "/context", "org.example.Context") + .unwrap() + .call::("Callback", ()) + .await; + assert!( + ordinary.is_err(), + "scoped confidential endpoints refuse ordinary calls even from their module" + ); + assert_eq!(calls.load(Ordering::SeqCst), 1); + let attacker = Connection::connect(bus.connect().await.unwrap()) + .await + .unwrap(); + let mut forged = crate::message::Message::method_call( + binding.unique_name().clone(), + "/callback".parse().unwrap(), + "org.example.Callback".parse().unwrap(), + "Observe".parse().unwrap(), + serde_json::json!([]), + ); + forged.header.sender = Some(module.clone()); + forged.header.confidential = true; + assert!( + attacker + .call_raw(forged, Duration::from_secs(2)) + .await + .is_err() + ); + assert_eq!(calls.load(Ordering::SeqCst), 1); + binding.revoke(); + let revoked = binding + .proxy(module.as_str(), "/context", "org.example.Context") + .unwrap() + .call_confidential::("ConfidentialCallback", ()) + .await; + assert!(revoked.is_err()); + assert_eq!(calls.load(Ordering::SeqCst), 1); + task.abort(); +} + +struct CountingNativeCallback(Arc); +#[async_trait::async_trait] +impl crate::Interface for CountingNativeCallback { + fn name(&self) -> crate::InterfaceName { + "org.example.Callback".parse().unwrap() + } + fn members(&self) -> Vec { + vec!["Observe".parse().unwrap()] + } + async fn call( + &self, + _: &crate::MemberName, + _: serde_json::Value, + ) -> crate::Result { + Err(Error::failed("missing contextual dispatch")) + } + async fn call_with_context( + &self, + _: &crate::MemberName, + _: serde_json::Value, + context: &crate::CallContext, + ) -> crate::Result { + let sender = context + .authenticated_sender() + .ok_or_else(|| Error::failed("unverified callback"))?; + self.0.fetch_add(1, Ordering::SeqCst); + Ok(serde_json::json!(sender)) + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +#[ignore = "requires TINYBUS_TEST_HOST_CALLBACK_MODULE to point at the built cdylib"] +async fn eager_native_setup_callbacks_are_bound_before_ready_and_failures_rollback() { + for failed in [false, true] { + let artifact = + PathBuf::from(std::env::var_os("TINYBUS_TEST_HOST_CALLBACK_MODULE").unwrap()); + let hash = crate::module::hash::file_hex(std::fs::File::open(&artifact).unwrap()).unwrap(); + let (_dir, artifact) = staged_module(&artifact, &hash); + let broker = Broker::new(); + let modules = ModuleHost::new(broker.clone()); + let binding = modules + .prepare_host_callbacks(vec![crate::module::HostCallbackEndpoint { + path: "/callback".parse().unwrap(), + interface: "org.example.Callback".parse().unwrap(), + members: vec!["Observe".parse().unwrap()], + }]) + .await + .unwrap(); + let calls = Arc::new(AtomicUsize::new(0)); + let guard = binding.guard(); + binding + .serve_at( + "/callback".parse().unwrap(), + BoundNativeCallback { + calls: calls.clone(), + guard: guard.clone(), + }, + ) + .await + .unwrap(); + let config = serde_json::json!({ + "host": binding.unique_name(), "fail": failed, + "credential": "callback-secret-must-not-be-retained", + "nested": [{"zero": -0.0, "literal": "\"-0.0\""}], + }); + let result = modules + .load_file_with_host_callbacks(artifact, config.clone(), &binding) + .await; + assert_eq!(calls.load(Ordering::SeqCst), usize::from(!failed)); + if failed { + assert!(result.is_err()); + assert!(modules.list().is_empty()); + assert!(!guard.authorizes(&crate::CallContext::brokered( + guard.module_peer().as_ref(), + true + ))); + } else { + let info = result.unwrap(); + assert!(matches!(info.state, ModuleState::Ready)); + assert_eq!( + Some(binding.unique_name().clone()), + broker + .router + .lock() + .unwrap() + .owner_of(binding.unique_name()) + ); + assert!( + modules + .reinitialize( + &info.name, + serde_json::json!({"host": ":1.999", "fail": false}) + ) + .await + .is_err() + ); + assert_eq!(calls.load(Ordering::SeqCst), 1); + let retains_plaintext = { + let loaded = modules.inner.loaded.lock().unwrap(); + let retained = &loaded + .iter() + .find(|m| m.info.name == info.name) + .unwrap() + .callback_config; + // Check the live retained representation, never freed memory after Drop. + (retained as &dyn std::any::Any) + .downcast_ref::>() + .is_some_and(Option::is_some) + }; + assert!( + !retains_plaintext, + "initialized callback module retains secret-bearing JSON" + ); + let mut equivalent = config.clone(); + equivalent["nested"][0]["zero"] = serde_json::json!(0.0); + assert_eq!(equivalent, config); + modules.reinitialize(&info.name, equivalent).await.unwrap(); + let mut different = config.clone(); + different["nested"][0]["zero"] = serde_json::json!(0); + assert_ne!(different, config); + assert!(modules.reinitialize(&info.name, different).await.is_err()); + modules.reinitialize(&info.name, config).await.unwrap(); + assert_eq!(calls.load(Ordering::SeqCst), 3); + drop(modules); + assert!(!guard.authorizes(&crate::CallContext::brokered( + guard.module_peer().as_ref(), + true + ))); + } + binding.close().await.unwrap(); + } +} + +struct BoundNativeCallback { + calls: Arc, + guard: crate::module::HostCallbackGuard, +} +#[async_trait::async_trait] +impl crate::Interface for BoundNativeCallback { + fn name(&self) -> crate::InterfaceName { + "org.example.Callback".parse().unwrap() + } + fn members(&self) -> Vec { + vec!["Observe".parse().unwrap()] + } + async fn call( + &self, + _: &crate::MemberName, + _: serde_json::Value, + ) -> crate::Result { + Err(Error::failed("unverified")) + } + async fn call_with_context( + &self, + _: &crate::MemberName, + _: serde_json::Value, + context: &crate::CallContext, + ) -> crate::Result { + assert_eq!( + context.authenticated_sender(), + self.guard.module_peer().as_ref() + ); + assert!( + self.guard.authorizes(context), + "peer must be pinned before eager callback" + ); + self.calls.fetch_add(1, Ordering::SeqCst); + Ok(serde_json::json!(context.authenticated_sender().unwrap())) + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +#[ignore = "requires TINYBUS_TEST_CONTEXT_MODULE to point at the built cdylib"] +async fn a_native_callback_timeout_and_revoke_do_not_claim_worker_completion() { + let artifact = PathBuf::from(std::env::var_os("TINYBUS_TEST_CONTEXT_MODULE").unwrap()); + let hash = crate::module::hash::file_hex(std::fs::File::open(&artifact).unwrap()).unwrap(); + let (_dir, artifact) = staged_module(&artifact, &hash); + let modules = ModuleHost::new(Broker::new()); + let binding = modules + .prepare_host_callbacks(vec![crate::module::HostCallbackEndpoint { + path: "/callback".parse().unwrap(), + interface: "org.example.Callback".parse().unwrap(), + members: vec!["Observe".parse().unwrap()], + }]) + .await + .unwrap(); + let guard = binding.guard(); + let (entered, mut receiver) = tokio::sync::mpsc::channel(1); + let release = Arc::new(tokio::sync::Semaphore::new(0)); + let published = Arc::new(AtomicUsize::new(0)); + binding + .serve_at( + "/callback".parse().unwrap(), + NativeBlockingCallback { + guard: guard.clone(), + entered, + release: release.clone(), + published: published.clone(), + }, + ) + .await + .unwrap(); + modules + .load_file_with_host_callbacks(artifact, serde_json::json!({}), &binding) + .await + .unwrap(); + let module = guard.module_peer().unwrap(); + let proxy = binding + .proxy(module.as_str(), "/context", "org.example.Context") + .unwrap() + .with_timeout(Duration::from_millis(100)); + let request = tokio::spawn(async move { + proxy + .call_confidential::("ConfidentialCallback", ()) + .await + }); + tokio::time::timeout(Duration::from_secs(2), receiver.recv()) + .await + .unwrap() + .unwrap(); + assert!(matches!(request.await.unwrap(), Err(Error::Timeout { .. }))); + assert_eq!( + guard.running(), + 1, + "caller timeout does not complete the host worker" + ); + binding.revoke(); + assert_eq!( + guard.running(), + 1, + "revoke retains running admission charge" + ); + release.add_permits(1); + tokio::time::timeout(Duration::from_secs(2), guard.drained()) + .await + .unwrap(); + assert_eq!(guard.running(), 0); + assert_eq!(published.load(Ordering::SeqCst), 0); + modules.shutdown(Duration::from_secs(2)).await; +} + +struct NativeBlockingCallback { + guard: crate::module::HostCallbackGuard, + entered: tokio::sync::mpsc::Sender<()>, + release: Arc, + published: Arc, +} +#[async_trait::async_trait] +impl crate::Interface for NativeBlockingCallback { + fn name(&self) -> crate::InterfaceName { + "org.example.Callback".parse().unwrap() + } + fn members(&self) -> Vec { + vec!["Observe".parse().unwrap()] + } + async fn call( + &self, + _: &crate::MemberName, + _: serde_json::Value, + ) -> crate::Result { + Err(Error::failed("unverified")) + } + async fn call_with_context( + &self, + _: &crate::MemberName, + _: serde_json::Value, + context: &crate::CallContext, + ) -> crate::Result { + assert!(self.guard.authorizes(context)); + self.entered.send(()).await.unwrap(); + let _permit = self.release.acquire().await.unwrap(); + if !self.guard.authorizes(context) { + return Err(Error::failed("revoked before publication")); + } + self.published.fetch_add(1, Ordering::SeqCst); + Ok(serde_json::json!(context.authenticated_sender().unwrap())) + } +} diff --git a/crates/tinybus/src/module/mod.rs b/crates/tinybus/src/module/mod.rs index 5d54bfeb..b06e2cd2 100644 --- a/crates/tinybus/src/module/mod.rs +++ b/crates/tinybus/src/module/mod.rs @@ -37,6 +37,10 @@ pub use first_admitted::{ReleaseAsset, ReleasePlan, load_first_admitted}; #[cfg(feature = "modules")] pub use github::CachedRelease; #[cfg(feature = "modules")] +pub use host::callbacks::{ + HostCallbackBinding, HostCallbackEndpoint, HostCallbackGuard, HostCallbackProxy, +}; +#[cfg(feature = "modules")] pub use host::{ModuleHost, ModuleInfo, ModuleState}; /// Entry points compiled into the host instead of loaded from a shared library. diff --git a/crates/tinybus/src/module/transport.rs b/crates/tinybus/src/module/transport.rs index d8b7d3a2..35097b42 100644 --- a/crates/tinybus/src/module/transport.rs +++ b/crates/tinybus/src/module/transport.rs @@ -57,6 +57,8 @@ pub(crate) struct ModuleTransport { // let a second lifecycle callback overlap the still-running first one. lifecycle: Arc>, stop_task_spawned: Arc, + initialization_gate: StdMutex>, + initialization_gate_changed: Notify, } // `module_ctx` is opaque and all access to it goes through callbacks whose ABI @@ -127,6 +129,8 @@ impl ModuleTransport { drain_started: AtomicBool::new(false), lifecycle: Arc::new(Mutex::new(())), stop_task_spawned: Arc::new(AtomicBool::new(false)), + initialization_gate: StdMutex::new(Some(true)), + initialization_gate_changed: Notify::new(), }); let config = context.config.lock().expect("module config lock"); let config_slice = crate::module::abi::TbSlice { @@ -186,7 +190,54 @@ impl ModuleTransport { }); } - async fn ensure_initialized(&self) -> Result<()> { + pub(crate) fn pause_initialization(&self) { + *self + .initialization_gate + .lock() + .expect("initialization gate lock") = None; + } + + pub(crate) fn release_initialization(&self, allowed: bool) { + *self + .initialization_gate + .lock() + .expect("initialization gate lock") = Some(allowed); + self.initialization_gate_changed.notify_waiters(); + if !allowed { + self.terminate_waiters(); + } + } + + // Terminal state is distinct from initialization: never fabricate READY or + // init_started to release observers during admission rollback or shutdown. + fn terminate_waiters(&self) { + self.context.faulted.store(true, Ordering::Release); + self.context.init_notify.notify_waiters(); + self.context.ready_notify.notify_waiters(); + self.initialization_gate_changed.notify_waiters(); + } + + pub(crate) async fn ensure_initialized(&self) -> Result<()> { + loop { + let changed = self.initialization_gate_changed.notified(); + let terminal = self.context.init_notify.notified(); + let allowed = *self + .initialization_gate + .lock() + .expect("initialization gate lock"); + if self.is_faulted() { + return Err(Error::ConnectionClosed); + } + match allowed { + Some(true) => break, + Some(false) => return Err(Error::transport("module callback admission failed")), + None => tokio::select! { + _ = changed => {}, + _ = terminal => {}, + }, + } + } + let result = self .init_result .get_or_init(|| async { @@ -244,7 +295,7 @@ impl ModuleTransport { } pub(crate) fn is_ready(&self) -> bool { - self.context.ready.load(Ordering::Acquire) + self.context.ready.load(Ordering::Acquire) && !self.is_faulted() } pub(crate) fn clear_config(&self) { @@ -254,7 +305,7 @@ impl ModuleTransport { config.shrink_to_fit(); } - pub(crate) async fn reinitialize(self: &Arc, config: serde_json::Value) -> Result<()> { + pub(crate) async fn reinitialize(self: &Arc, bytes: crate::Secret) -> Result<()> { let started = std::time::Instant::now(); let lifecycle = tokio::time::timeout(MODULE_INIT_DEADLINE, self.lifecycle.clone().lock_owned()) @@ -276,10 +327,6 @@ impl ModuleTransport { .reinitialize .ok_or_else(|| Error::failed("module does not support reinitialization"))?; let module = SendModuleVtable(module); - let bytes = crate::Secret::new( - serde_json::to_vec(&config) - .map_err(|_| Error::failed("module configuration is invalid"))?, - ); let remaining = MODULE_INIT_DEADLINE.saturating_sub(started.elapsed()); let task = tokio::task::spawn_blocking(move || { let _lifecycle = lifecycle; @@ -326,7 +373,7 @@ impl ModuleTransport { pub(crate) async fn wait_initializing(&self) { loop { let notified = self.context.init_notify.notified(); - if self.init_started() { + if self.init_started() || self.is_faulted() { return; } notified.await; @@ -470,6 +517,7 @@ impl ModuleTransport { } pub(crate) fn stop_sync(&self, deadline: Duration) -> i32 { + self.terminate_waiters(); let code = self.shutdown_sync(deadline); self.context .inbound @@ -681,6 +729,7 @@ unsafe extern "C" fn host_fault(ctx: *mut c_void, _: *const u8, _: usize) { let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { if let Some(context) = unsafe { ctx.cast::().as_ref() } { context.faulted.store(true, Ordering::Release); + context.init_notify.notify_waiters(); context.ready_notify.notify_waiters(); context.inbound.lock().expect("host inbound lock").take(); } diff --git a/crates/tinybus/src/module/transport_tests.rs b/crates/tinybus/src/module/transport_tests.rs index 8ef74311..cee6a353 100644 --- a/crates/tinybus/src/module/transport_tests.rs +++ b/crates/tinybus/src/module/transport_tests.rs @@ -122,7 +122,9 @@ async fn an_older_vtable_loads_but_reports_reinitialization_as_unsupported() { older.size = TB_MODULE_VTABLE_BASE_SIZE; transport.initialize(older).unwrap(); let error = transport - .reinitialize(serde_json::json!({ "secret": "never printed" })) + .reinitialize(crate::Secret::new( + serde_json::to_vec(&serde_json::json!({ "secret": "never printed" })).unwrap(), + )) .await .unwrap_err(); assert!(error.to_string().contains("does not support")); @@ -151,7 +153,9 @@ async fn reinitialization_maps_callback_failures_without_exposing_configuration( let (uninitialized, _) = ModuleTransport::new("missing".to_string(), Vec::new()); assert!( uninitialized - .reinitialize(serde_json::json!({})) + .reinitialize(crate::Secret::new( + serde_json::to_vec(&serde_json::json!({})).unwrap() + )) .await .unwrap_err() .to_string() @@ -170,7 +174,9 @@ async fn reinitialization_maps_callback_failures_without_exposing_configuration( ] { REINITIALIZE_CODE.store(code, Ordering::Release); let error = transport - .reinitialize(serde_json::json!({ "secret": "never printed" })) + .reinitialize(crate::Secret::new( + serde_json::to_vec(&serde_json::json!({ "secret": "never printed" })).unwrap(), + )) .await .unwrap_err(); assert!(error.to_string().contains(expected), "{error}"); @@ -178,7 +184,9 @@ async fn reinitialization_maps_callback_failures_without_exposing_configuration( } REINITIALIZE_CODE.store(TB_OK, Ordering::Release); transport - .reinitialize(serde_json::json!({ "replacement": true })) + .reinitialize(crate::Secret::new( + serde_json::to_vec(&serde_json::json!({ "replacement": true })).unwrap(), + )) .await .unwrap(); } @@ -201,7 +209,9 @@ async fn a_running_reinitialization_excludes_another_reinitialization_and_shutdo let first_transport = transport.clone(); let first = tokio::spawn(async move { first_transport - .reinitialize(serde_json::json!({ "generation": 1 })) + .reinitialize(crate::Secret::new( + serde_json::to_vec(&serde_json::json!({ "generation": 1 })).unwrap(), + )) .await }); tokio::time::timeout(Duration::from_secs(1), async { @@ -215,7 +225,9 @@ async fn a_running_reinitialization_excludes_another_reinitialization_and_shutdo assert!( tokio::time::timeout( Duration::from_millis(20), - transport.reinitialize(serde_json::json!({ "generation": 2 })), + transport.reinitialize(crate::Secret::new( + serde_json::to_vec(&serde_json::json!({ "generation": 2 })).unwrap() + )), ) .await .is_err() @@ -438,7 +450,9 @@ async fn a_reinitialization_that_times_out_reports_a_timeout_without_the_configu transport.initialize(module).unwrap(); REINITIALIZE_CODE.store(TB_TIMEOUT, Ordering::Release); let error = transport - .reinitialize(serde_json::json!({ "secret": "never printed" })) + .reinitialize(crate::Secret::new( + serde_json::to_vec(&serde_json::json!({ "secret": "never printed" })).unwrap(), + )) .await .unwrap_err(); REINITIALIZE_CODE.store(TB_OK, Ordering::Release); @@ -524,3 +538,56 @@ async fn a_new_host_keeps_the_frozen_prefix_readable_by_old_modules() { ); transport.initialize(module).unwrap(); } + +#[tokio::test] +async fn rejected_initialization_releases_observer_without_claiming_initializing() { + let (transport, _) = ModuleTransport::new("rejected".into(), vec![]); + transport.pause_initialization(); + let weak = Arc::downgrade(&transport); + let observed = transport.clone(); + let observer = tokio::spawn(async move { observed.wait_initializing().await }); + tokio::task::yield_now().await; + transport.release_initialization(false); + assert!(transport.ensure_initialized().await.is_err()); + assert!(!transport.init_started()); + drop(transport); + tokio::time::timeout(Duration::from_millis(100), observer) + .await + .unwrap() + .unwrap(); + assert!(weak.upgrade().is_none()); +} + +#[tokio::test] +async fn stop_before_initialization_terminates_all_waiters() { + let (transport, _) = ModuleTransport::new("stopped".into(), vec![]); + transport.pause_initialization(); + transport.stop_sync(Duration::ZERO); + tokio::time::timeout(Duration::from_millis(100), transport.ensure_initialized()) + .await + .unwrap() + .unwrap_err(); + tokio::time::timeout(Duration::from_millis(100), transport.wait_initializing()) + .await + .unwrap(); + tokio::time::timeout(Duration::from_millis(100), transport.wait_ready()) + .await + .unwrap(); + assert!(!transport.init_started()); + assert!(!transport.is_ready()); +} + +#[tokio::test] +async fn ready_followed_by_fault_is_not_successful_readiness() { + let (transport, host) = ModuleTransport::new("ready-fault".into(), vec![]); + unsafe { + (host.ready)(host.host_ctx); + } + assert!(transport.is_ready()); + unsafe { + (host.fault)(host.host_ctx, std::ptr::null(), 0); + } + transport.wait_ready().await; + assert!(transport.is_faulted()); + assert!(!transport.is_ready()); +} diff --git a/crates/tinybus/src/proxy.rs b/crates/tinybus/src/proxy.rs index 9e7cd9a5..367790ed 100644 --- a/crates/tinybus/src/proxy.rs +++ b/crates/tinybus/src/proxy.rs @@ -129,12 +129,16 @@ impl Proxy { /// /// The call fails with [`crate::Error::NotAttested`] unless the broker has /// itself verified the destination's artifact against the operator's trust - /// store. That refusal is the feature: handing a private key to whoever + /// store, or this exact call matches an active SDK-created local host callback + /// capability for this verified loaded module sender. That refusal is the feature: handing a private key to whoever /// happened to claim the name first is the outcome this exists to prevent, /// and it is better to fail a deploy than to succeed at that. /// - /// Use [`Proxy::attestation`] first if the caller wants to distinguish - /// "not installed" from "not trusted" before it assembles the secret. + /// For artifact-backed modules, use [`Proxy::attestation`] first to + /// distinguish "not installed" from "not trusted" before assembling a secret. + /// For scoped host callbacks, use the SDK-created binding and its scope + /// checks: callback authority comes from that local ModuleHost capability, + /// which deliberately has no artifact attestation. pub async fn call_confidential( &self, member: &str, diff --git a/crates/tinybus/src/router.rs b/crates/tinybus/src/router.rs index 6e2fddc5..0027c07a 100644 --- a/crates/tinybus/src/router.rs +++ b/crates/tinybus/src/router.rs @@ -24,6 +24,8 @@ //! to discard a signal it never asked for is exactly that cost in miniature. use std::collections::HashMap; +#[cfg(feature = "modules")] +use std::sync::Arc; use tokio::sync::mpsc; @@ -204,6 +206,8 @@ pub(crate) struct Router { peers: HashMap, names: HashMap, next_id: u64, + #[cfg(feature = "modules")] + host_callbacks: Vec>, } /// What changed when a name's owner changed, so the broker can announce it. @@ -246,6 +250,15 @@ impl Router { let Some(peer) = self.peers.remove(&id) else { return Vec::new(); }; + #[cfg(feature = "modules")] + self.host_callbacks.retain(|capability| { + if capability.includes_peer(&peer.unique) { + capability.revoke(); + false + } else { + true + } + }); let owned: Vec = self .names .iter() @@ -266,6 +279,25 @@ impl Router { changes } + #[cfg(feature = "modules")] + pub(crate) fn detach_unique(&mut self, unique: &BusName) { + if let Some(id) = self.names.get(unique).copied() { + self.detach(id); + } + } + + #[cfg(feature = "modules")] + pub(crate) fn revoke_callbacks_for_peer(&mut self, peer: &BusName) { + self.host_callbacks.retain(|entry| { + if entry.includes_peer(peer) { + entry.revoke(); + false + } else { + true + } + }); + } + /// Claim a well-known name for `id`. /// /// First writer wins and there is no queue: a second claimant is told the @@ -490,17 +522,63 @@ impl Router { destination: &BusName, expected: Option<&BusName>, confidential: bool, + #[cfg_attr(not(feature = "modules"), allow(unused_variables))] + header: &crate::message::Header, ) -> Result> { if expected.is_some_and(|expected| self.owner_of(destination).as_ref() != Some(expected)) { return Err(Error::NameHasNoOwner(destination.clone())); } if confidential { + #[cfg(feature = "modules")] + if destination.is_unique() + && self + .host_callbacks + .iter() + .any(|capability| capability.matches(header)) + { + return self.resolve(destination); + } self.resolve_attested(destination) } else { self.resolve(destination) } } + #[cfg(feature = "modules")] + pub(crate) fn add_host_callbacks( + &mut self, + capability: Arc, + ) -> Result<()> { + if self.host_callbacks.iter().any(|entry| { + Arc::ptr_eq(entry, &capability) || entry.module_peer() == capability.module_peer() + }) { + return Err(Error::failed("callback scope is already registered")); + } + if self.host_callbacks.len() >= 128 { + return Err(Error::failed("callback scope limit reached")); + } + let source = capability + .module_peer() + .ok_or_else(|| Error::failed("callback module peer is absent"))?; + if self.attestation_of(&source).is_none() + || self.owner_of(&capability.destination).is_none() + { + return Err(Error::failed("callback peers are not admitted")); + } + self.host_callbacks.push(capability); + Ok(()) + } + + #[cfg(feature = "modules")] + pub(crate) fn revoke_host_callbacks( + &mut self, + capability: &Arc, + ) { + capability.revoke(); + self.host_callbacks + .retain(|entry| !Arc::ptr_eq(entry, capability)); + } + /// The outbox of whoever owns `destination`. pub fn resolve(&self, destination: &BusName) -> Result> { let id = self diff --git a/crates/tinybus/src/service/mod.rs b/crates/tinybus/src/service/mod.rs index 07ade4c1..807b5b9d 100644 --- a/crates/tinybus/src/service/mod.rs +++ b/crates/tinybus/src/service/mod.rs @@ -48,6 +48,7 @@ pub use crate::service::tree::ObjectTree; #[derive(Debug, Clone, Default)] pub struct CallContext { sender: Option, + confidential: bool, } impl CallContext { @@ -60,9 +61,16 @@ impl CallContext { self.sender.as_ref() } - pub(crate) fn brokered(sender: Option<&BusName>) -> Self { + /// Whether the authenticated incoming call requested confidential routing. + /// Unverified contexts never assert confidential admission. + pub fn is_confidential(&self) -> bool { + self.confidential + } + + pub(crate) fn brokered(sender: Option<&BusName>, confidential: bool) -> Self { Self { sender: sender.filter(|name| name.is_unique()).cloned(), + confidential: confidential && sender.is_some_and(BusName::is_unique), } } } diff --git a/crates/tinybus/src/service/mod_tests.rs b/crates/tinybus/src/service/mod_tests.rs index 32881059..b0d346de 100644 --- a/crates/tinybus/src/service/mod_tests.rs +++ b/crates/tinybus/src/service/mod_tests.rs @@ -3,16 +3,20 @@ use super::*; #[test] fn only_a_brokered_unique_sender_is_authenticated() { assert!(CallContext::default().authenticated_sender().is_none()); - assert!(CallContext::brokered(None).authenticated_sender().is_none()); + assert!( + CallContext::brokered(None, false) + .authenticated_sender() + .is_none() + ); let well_known = BusName::new("org.example.Host").unwrap(); assert!( - CallContext::brokered(Some(&well_known)) + CallContext::brokered(Some(&well_known), false) .authenticated_sender() .is_none() ); let unique = BusName::new(":1.42").unwrap(); assert_eq!( - CallContext::brokered(Some(&unique)).authenticated_sender(), + CallContext::brokered(Some(&unique), false).authenticated_sender(), Some(&unique) ); } diff --git a/crates/tinybus/src/transport/memory.rs b/crates/tinybus/src/transport/memory.rs index 32e88188..89fa6a63 100644 --- a/crates/tinybus/src/transport/memory.rs +++ b/crates/tinybus/src/transport/memory.rs @@ -93,6 +93,11 @@ impl MemoryTransport { .map(|delivery| (delivery.message, delivery.brokered))) } + #[cfg(feature = "modules")] + pub(crate) fn close_outbound(&self) { + self.outbound.lock().expect("outbound lock").take(); + } + /// Take a clone of the sender without holding the lock across an await. fn sender(&self) -> Result> { self.outbound diff --git a/docs/modules/attest/README.md b/docs/modules/attest/README.md index 5d36a3ac..ade189ac 100644 --- a/docs/modules/attest/README.md +++ b/docs/modules/attest/README.md @@ -5,12 +5,14 @@ delivered. ## Who receives a secret -A **module loaded into the host's address space**, and nothing else. +A **verified module loaded into the host's address space**, or an explicitly +scoped local host callback created by `ModuleHost`. The callback exception binds +one verified loaded sender to one SDK-created local connection and exact +path/interface/member endpoints. It never produces an artifact attestation. +See [scoped local host callbacks](#scoped-local-host-callbacks). -That is the whole rule, and it is a deliberate narrowing rather than a -limitation. A secret handed to a loaded module never crosses a transport, never -reaches a separate process, and never touches a socket — so there is no peer to -identify, no process credential to read, and no per-OS code to keep working. +Neither recipient crosses a process boundary or a socket. Ordinary connections, +remote peers and well-known host aliases cannot acquire this callback permission. Services in their own processes, CLI clients and monitors do not receive secrets. A confidential message addressed to one is refused. @@ -186,3 +188,43 @@ TINYBUS_TEST_MODULE="$PWD/target/debug/examples/libmodule_clock.so" \ The artifact name above is Linux's (`libmodule_clock.so`); on macOS Cargo builds `libmodule_clock.dylib` instead, so point `TINYBUS_TEST_MODULE` at that file when running the same command there. + + +## Scoped local host callbacks + +`ModuleHost::prepare_host_callbacks` creates its own local MemoryTransport +connection. It accepts at most 32 exact path/interface declarations, each with +one to 64 distinct members. `HostCallbackBinding::serve_at` wraps those methods +in authenticated, confidential, live dispatch checks; it exposes no raw +connection for bypassing that guard. An outgoing `proxy` uses this same private +host peer, so initialization and callback reception share one fixed identity. + +`load_file_with_host_callbacks` requires an actual hash-verified local artifact +and all endpoints served before admission. It reserves the exact loaded module +peer and binds the exact sender, unique host destination, path, interface and +member before releasing its initialization gate. The read-only `guard()` exposes +`module_peer()` before eager setup callbacks run. The loader awaits actual +READY/fault with a deadline; ABI initialization success alone is insufficient. +Failures or a dropped load future revoke routes and detach reserved peers. A +binding is single-use. Reinitialization requires its unchanged configuration. + +This intentional recipient exception is **not artifact attestation**. There is +no remote enrollment method, self-attestation, wire trust flag, existing +connection registration, alias inheritance or generic-load callback grant. The +router admits at most 128 active scopes and continues to apply ordinary module +artifact admission and expected-owner checks. + +`revoke()` is immediate and idempotent for new route and user-dispatch admission. +Drop, peer detach and host-owner drop also revoke. A callback selected earlier +must recheck liveness before user code. Code already running remains counted by +`guard().running()` until its future completes or drops. `drained()` reports +that count only; caller timeouts and connection close do not imply that provider +writes have ceased. Detached provider tasks require their own completion fence. +Domain callbacks must check `guard().authorizes(context)` after awaits and before +publication and preserve uncertainty about writes already attempted. No buffer +copy wiping or whole-boundary zeroization guarantee is made. + +`module_host_callback` is an actual cdylib fixture: hash-verified bytes perform +an eager confidential callback during setup before READY. The native-loader CI +job runs it on Linux, macOS and Windows. `module_context` separately proves +confidential runtime callbacks and refusal of ordinary, forged and revoked calls. diff --git a/scripts/check-coverage.sh b/scripts/check-coverage.sh index 911b92c9..2e49d32b 100755 --- a/scripts/check-coverage.sh +++ b/scripts/check-coverage.sh @@ -1,14 +1,15 @@ #!/usr/bin/env bash set -euo pipefail cd "$(dirname "${BASH_SOURCE[0]}")/.." -cargo build --locked --example module_clock --example module_clock_two --example module_wrong_target --example module_context --example module_context_eager --no-default-features --features modules,macros +cargo build --locked --example module_clock --example module_clock_two --example module_wrong_target --example module_context --example module_context_eager --example module_host_callback --no-default-features --features modules,macros set -euo pipefail module=$(find target/debug/examples -maxdepth 1 -type f -name 'libmodule_clock.so' -print -quit) module_two=$(find target/debug/examples -maxdepth 1 -type f -name 'libmodule_clock_two.so' -print -quit) wrong_target=$(find target/debug/examples -maxdepth 1 -type f -name 'libmodule_wrong_target.so' -print -quit) context_module=$(find target/debug/examples -maxdepth 1 -type f -name 'libmodule_context.so' -print -quit) context_eager_module=$(find target/debug/examples -maxdepth 1 -type f -name 'libmodule_context_eager.so' -print -quit) -for artifact in "$module" "$module_two" "$wrong_target" "$context_module" "$context_eager_module"; do +host_callback_module=$(find target/debug/examples -maxdepth 1 -type f \( -name 'libmodule_host_callback.so' -o -name 'libmodule_host_callback.dylib' \) -print -quit) +for artifact in "$module" "$module_two" "$wrong_target" "$context_module" "$context_eager_module" "$host_callback_module"; do if [ -z "$artifact" ] || [ ! -f "$artifact" ]; then echo "a module loader fixture is missing" >&2 exit 1 @@ -32,4 +33,4 @@ printf '"%s" = "%s"\n' "$library" "$digest" > "$private/modules.toml" # process/network branches are not meaningful per-file coverage. The # test-support module is integration-test glue whose platform-specific # dynamic-loader and timeout paths run in the dedicated OS jobs instead. -TINYBUS_TEST_MODULE="$PWD/$module" TINYBUS_TEST_MODULE_TWO="$private/$library" TINYBUS_TEST_WRONG_TARGET="$PWD/$wrong_target" TINYBUS_TEST_CONTEXT_MODULE="$PWD/$context_module" TINYBUS_TEST_CONTEXT_EAGER_MODULE="$PWD/$context_eager_module" cargo llvm-cov --workspace --all-features --ignore-filename-regex 'crates/tinybus/src/ports/(listener|transport)\.rs$|crates/tinybus/src/module/github\.rs$|crates/tinybus/src/module/(host|mod)\.rs$|crates/tinybus/src/bin/(tinybus\.rs|tinybus/unix\.rs)$|crates/tinybus/src/connection\.rs$|crates/tinybus/src/test_support\.rs$' --fail-under-file-lines 90 -- --include-ignored +TINYBUS_TEST_MODULE="$PWD/$module" TINYBUS_TEST_MODULE_TWO="$private/$library" TINYBUS_TEST_WRONG_TARGET="$PWD/$wrong_target" TINYBUS_TEST_CONTEXT_MODULE="$PWD/$context_module" TINYBUS_TEST_CONTEXT_EAGER_MODULE="$PWD/$context_eager_module" TINYBUS_TEST_HOST_CALLBACK_MODULE="$PWD/$host_callback_module" cargo llvm-cov --workspace --all-features --ignore-filename-regex 'crates/tinybus/src/ports/(listener|transport)\.rs$|crates/tinybus/src/module/github\.rs$|crates/tinybus/src/module/(host|mod)\.rs$|crates/tinybus/src/bin/(tinybus\.rs|tinybus/unix\.rs)$|crates/tinybus/src/connection\.rs$|crates/tinybus/src/test_support\.rs$' --fail-under-file-lines 90 -- --include-ignored diff --git a/scripts/test-e2e.ps1 b/scripts/test-e2e.ps1 index a496449d..90592413 100644 --- a/scripts/test-e2e.ps1 +++ b/scripts/test-e2e.ps1 @@ -18,10 +18,11 @@ Set-Acl -LiteralPath $private -AclObject $acl Copy-Item target/debug/examples/module_clock.dll $private Copy-Item target/debug/examples/module_context.dll $private Copy-Item target/debug/examples/module_context_eager.dll $private +Copy-Item target/debug/examples/module_host_callback.dll $private Copy-Item target/debug/examples/module_wrong_target.dll $private # Admit every intentional fixture by its actual built bytes so later # checks, including the wrong-target gate, can exercise their contract. -$fixtureAllowlist = @('module_clock.dll', 'module_context.dll', 'module_context_eager.dll', 'module_wrong_target.dll') | ForEach-Object { +$fixtureAllowlist = @('module_clock.dll', 'module_context.dll', 'module_context_eager.dll', 'module_host_callback.dll', 'module_wrong_target.dll') | ForEach-Object { $digest = (Get-FileHash -LiteralPath (Join-Path $private $_) -Algorithm SHA256).Hash.ToLowerInvariant() '"' + $_ + '" = "' + $digest + '"' } @@ -44,6 +45,7 @@ $env:TEMP = $private $env:TMP = $private $env:TINYBUS_TEST_CONTEXT_MODULE = (Get-Item "$private/module_context.dll").FullName $env:TINYBUS_TEST_CONTEXT_EAGER_MODULE = (Get-Item "$private/module_context_eager.dll").FullName +$env:TINYBUS_TEST_HOST_CALLBACK_MODULE = (Get-Item "$private/module_host_callback.dll").FullName $env:TINYBUS_TEST_MODULE = $module $env:TINYBUS_TEST_MODULE_TWO = $moduleTwo $env:TINYBUS_TEST_WRONG_TARGET = $wrongTarget diff --git a/scripts/test-e2e.sh b/scripts/test-e2e.sh index b3010fb4..bf97e755 100755 --- a/scripts/test-e2e.sh +++ b/scripts/test-e2e.sh @@ -1,7 +1,7 @@ #!/usr/bin/env bash set -euo pipefail cd "$(dirname "${BASH_SOURCE[0]}")/.." -cargo build --locked --example module_clock --example module_clock_two --example module_wrong_target --example module_context --example module_context_eager --no-default-features --features modules,macros +cargo build --locked --example module_clock --example module_clock_two --example module_wrong_target --example module_context --example module_context_eager --example module_host_callback --no-default-features --features modules,macros cargo test --locked --no-default-features --features modules,macros,test-support test_support::tests case "${RUNNER_OS:-$(uname -s)}" in Windows|MINGW*|MSYS*|CYGWIN*) pwsh -NoProfile -File scripts/test-e2e.ps1 ;; @@ -12,7 +12,8 @@ module_two=$(find target/debug/examples -maxdepth 1 -type f \( -name 'libmodule_ wrong_target=$(find target/debug/examples -maxdepth 1 -type f \( -name 'libmodule_wrong_target.so' -o -name 'libmodule_wrong_target.dylib' \) -print -quit) context_module=$(find target/debug/examples -maxdepth 1 -type f \( -name 'libmodule_context.so' -o -name 'libmodule_context.dylib' \) -print -quit) context_eager_module=$(find target/debug/examples -maxdepth 1 -type f \( -name 'libmodule_context_eager.so' -o -name 'libmodule_context_eager.dylib' \) -print -quit) -for artifact in "$module" "$module_two" "$wrong_target" "$context_module" "$context_eager_module"; do +host_callback_module=$(find target/debug/examples -maxdepth 1 -type f \( -name 'libmodule_host_callback.so' -o -name 'libmodule_host_callback.dylib' \) -print -quit) +for artifact in "$module" "$module_two" "$wrong_target" "$context_module" "$context_eager_module" "$host_callback_module"; do if [ -z "$artifact" ] || [ ! -f "$artifact" ]; then echo "a module example artifact is missing; the loader tests would not exercise it" >&2 exit 1 @@ -29,7 +30,7 @@ else digest="$(shasum -a 256 "$private/$library" | awk '{print $1}')" fi printf '"%s" = "%s"\n' "$library" "$digest" > "$private/modules.toml" -TINYBUS_TEST_MODULE="$PWD/$module" TINYBUS_TEST_MODULE_TWO="$private/$library" TINYBUS_TEST_WRONG_TARGET="$PWD/$wrong_target" TINYBUS_TEST_CONTEXT_MODULE="$PWD/$context_module" TINYBUS_TEST_CONTEXT_EAGER_MODULE="$PWD/$context_eager_module" cargo test --locked --no-default-features --features modules,macros,test-support -- --ignored +TINYBUS_TEST_MODULE="$PWD/$module" TINYBUS_TEST_MODULE_TWO="$private/$library" TINYBUS_TEST_WRONG_TARGET="$PWD/$wrong_target" TINYBUS_TEST_CONTEXT_MODULE="$PWD/$context_module" TINYBUS_TEST_CONTEXT_EAGER_MODULE="$PWD/$context_eager_module" TINYBUS_TEST_HOST_CALLBACK_MODULE="$PWD/$host_callback_module" cargo test --locked --no-default-features --features modules,macros,test-support -- --ignored cargo test --locked --no-default-features --features modules,macros admission_tests -- --include-ignored TINYBUS_TEST_UMASK=1 cargo test --locked --no-default-features --features modules,macros permissive_umask -- --test-threads=1