Add confined native module callbacks to ordinary hosts

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

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"]

crates/tinybus/examples/module_context.rs

@@ -12,7 +12,11 @@ impl Interface for ContextService {
         "org.example.Context".parse().unwrap()
     }
     fn members(&self) -> Vec<MemberName> {
-        vec!["Inspect".parse().unwrap(), "Callback".parse().unwrap()]
+        vec![
+            "Inspect".parse().unwrap(),
+            "Callback".parse().unwrap(),
+            "ConfidentialCallback".parse().unwrap(),
+        ]
     }
     async fn call(&self, _: &MemberName, _: Value) -> Result<Value> {
         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,
 }

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,
+}

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.
 //!

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<Mutex<Router>>,
+    pub(crate) router: Arc<Mutex<Router>>,
     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| {

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()
     };

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<String>,
-    /// 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

crates/tinybus/src/module/host.rs

@@ -96,6 +96,40 @@ struct LoadedModule {
     transition_from: Option<ModuleState>,
     descriptor_info: Option<DescriptorInfo>,
     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,13 +1523,26 @@ impl ModuleControl for ModuleHostInner {
     }
 
     async fn reinitialize(&self, name: &str, config: serde_json::Value) -> Result<ModuleInfo> {
+        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
                 .iter_mut()
                 .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;