From 0bbcecee375214ffe2a77018d219cdce8236f399 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 11 Oct 2026 19:12:03 +0300 Subject: [PATCH 01/14] chore: files changed crates/tinybus/src/connection.rs,crates/tinybus/src/connection_reply_tests.rs,t Auto-committed-on: dragonfly Co-authored-by: Medulla --- crates/tinybus/src/connection.rs | 4 ++ crates/tinybus/src/connection_reply_tests.rs | 43 +++++++++++++ target-forged-reply-red.log | 68 ++++++++++++++++++++ 3 files changed, 115 insertions(+) create mode 100644 crates/tinybus/src/connection_reply_tests.rs create mode 100644 target-forged-reply-red.log diff --git a/crates/tinybus/src/connection.rs b/crates/tinybus/src/connection.rs index 5aa246a4..df3e5632 100644 --- a/crates/tinybus/src/connection.rs +++ b/crates/tinybus/src/connection.rs @@ -1389,3 +1389,7 @@ async fn dispatch(inner: &Inner, header: &Header, body: Value, brokered: bool) - #[cfg(test)] #[path = "connection_tests.rs"] mod tests; + +#[cfg(test)] +#[path = "connection_reply_tests.rs"] +mod reply_tests; diff --git a/crates/tinybus/src/connection_reply_tests.rs b/crates/tinybus/src/connection_reply_tests.rs new file mode 100644 index 00000000..72dd244e --- /dev/null +++ b/crates/tinybus/src/connection_reply_tests.rs @@ -0,0 +1,43 @@ +use super::*; +use crate::broker::Broker; +use crate::transport::memory::MemoryBus; + +#[tokio::test] +async fn another_broker_peer_cannot_complete_a_waiter_with_a_forged_reply() { + let bus = MemoryBus::new(); + Broker::new().spawn(bus.clone()); + let responder = bus.connect().await.unwrap(); + let mut hello = Message::method_call( + BusName::new(crate::BUS_NAME).unwrap(), + ObjectPath::new(crate::BUS_PATH).unwrap(), + InterfaceName::new(crate::BUS_INTERFACE).unwrap(), + MemberName::new("Hello").unwrap(), serde_json::json!([]), + ); + hello.header.serial = 1; + responder.send(hello).await.unwrap(); + let responder_name = BusName::new(responder.recv().await.unwrap().unwrap().body.as_str().unwrap()).unwrap(); + let requester = Connection::connect(bus.connect().await.unwrap()).await.unwrap(); + let forger = Connection::connect(bus.connect().await.unwrap()).await.unwrap(); + let mut barriers = requester.add_match(MatchRule::new().signals()).await.unwrap(); + let request = Message::method_call( + responder_name.clone(), ObjectPath::new("/fixture").unwrap(), + InterfaceName::new("ai.tinyhumans.Fixture").unwrap(), + MemberName::new("Await").unwrap(), serde_json::json!([]), + ); + let pending = tokio::spawn({ let requester = requester.clone(); async move { requester.call_raw(request, DEFAULT_TIMEOUT).await } }); + let delivered = responder.recv().await.unwrap().unwrap(); + assert_eq!(delivered.header.sender, requester.unique_name()); + let mut forged = Message::method_return(&delivered.header, serde_json::json!("forged")); + // Even a claimed B sender is overwritten to the actual peer C. + forged.header.sender = Some(responder_name.clone()); + forger.send(forged).await.unwrap(); + forger.emit(ObjectPath::new("/fixture").unwrap(), InterfaceName::new("ai.tinyhumans.Fixture").unwrap(), MemberName::new("Barrier").unwrap(), ()).await.unwrap(); + let barrier = tokio::time::timeout(DEFAULT_TIMEOUT, barriers.recv()).await.unwrap().unwrap(); + assert_eq!(barrier.header.sender, forger.unique_name()); + // That connection's ordered signal arrived after its forged response, so + // requester dispatch has processed the forgery before B is allowed to reply. + let mut genuine = Message::method_return(&delivered.header, serde_json::json!("genuine")); + genuine.header.serial = 2; + responder.send(genuine).await.unwrap(); + assert_eq!(pending.await.unwrap().unwrap(), serde_json::json!("genuine")); +} diff --git a/target-forged-reply-red.log b/target-forged-reply-red.log new file mode 100644 index 00000000..d4f692a3 --- /dev/null +++ b/target-forged-reply-red.log @@ -0,0 +1,68 @@ +[ci-cancel-aware] exec: cargo test -p tinybus another_broker_peer_cannot_complete + Compiling proc-macro2 v1.0.107 + Compiling quote v1.0.47 + Compiling unicode-ident v1.0.24 + Compiling libc v0.2.189 + Compiling version_check v0.9.5 + Compiling typenum v1.20.1 + Compiling generic-array v0.14.7 + Compiling serde_core v1.0.229 + Compiling cfg-if v1.0.4 + Compiling once_cell v1.21.4 + Compiling zmij v1.0.23 + Compiling thiserror v2.0.20 + Compiling serde_json v1.0.151 + Compiling block-buffer v0.10.4 + Compiling crypto-common v0.1.7 + Compiling pin-project-lite v0.2.17 + Compiling serde v1.0.229 + Compiling digest v0.10.7 + Compiling syn v3.0.3 + Compiling syn v2.0.119 + Compiling tracing-core v0.1.36 + Compiling socket2 v0.6.5 + Compiling thiserror-impl v2.0.20 + Compiling tracing-attributes v0.1.31 + Compiling tokio-macros v2.7.2 + Compiling serde_derive v1.0.229 + Compiling mio v1.2.2 + Compiling itoa v1.0.18 + Compiling cpufeatures v0.2.17 + Compiling rustix v1.1.4 + Compiling memchr v2.8.3 + Compiling tinybus v0.1.3 (/home/enamakel/work/openhuman/worktrees/cloud-storage-7419/vendor/tinybus/crates/tinybus) + Compiling getrandom v0.4.3 + Compiling bytes v1.12.1 + Compiling sha2 v0.10.9 + Compiling tokio v1.53.1 + Compiling tracing v0.1.44 + Compiling tinybus-macros v0.1.3 (/home/enamakel/work/openhuman/worktrees/cloud-storage-7419/vendor/tinybus/crates/tinybus-macros) + Compiling async-trait v0.1.92 + Compiling linux-raw-sys v0.12.1 + Compiling bitflags v2.13.1 + Compiling fastrand v2.5.0 + Compiling tempfile v3.27.0 + Compiling tinybus-module v0.1.3 (/home/enamakel/work/openhuman/worktrees/cloud-storage-7419/vendor/tinybus/crates/tinybus-module) + Finished `test` profile [unoptimized + debuginfo] target(s) in 28.09s + Running unittests src/lib.rs (target/debug/deps/tinybus-49c252397df0a704) + +running 1 test +test connection::reply_tests::another_broker_peer_cannot_complete_a_waiter_with_a_forged_reply ... FAILED + +failures: + +---- connection::reply_tests::another_broker_peer_cannot_complete_a_waiter_with_a_forged_reply stdout ---- + +thread 'connection::reply_tests::another_broker_peer_cannot_complete_a_waiter_with_a_forged_reply' (3898707) panicked at crates/tinybus/src/connection_reply_tests.rs:42:5: +assertion `left == right` failed + left: String("forged") + right: String("genuine") +note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace + + +failures: + connection::reply_tests::another_broker_peer_cannot_complete_a_waiter_with_a_forged_reply + +test result: FAILED. 0 passed; 1 failed; 0 ignored; 0 measured; 282 filtered out; finished in 0.00s + +error: test failed, to rerun pass `-p tinybus --lib` From be653fa765ea4f5f622e457dabe836c6bf0bf8c5 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 11 Oct 2026 19:19:45 +0300 Subject: [PATCH 02/14] feat(tinybus): pin call replies to an expected responder Method calls can now carry an expected responder, and the broker resolves the destination and verifies that unique owner under a single routing lock before delivery. Clients capture the reply identity before sending and reject replies from any other sender, so a well-known name that changes hands mid-call can no longer redirect the answer. Auto-committed-on: dragonfly Co-authored-by: Medulla --- crates/tinybus/src/broker.rs | 24 ++++++----- crates/tinybus/src/connection.rs | 71 ++++++++++++++++++++++++++++--- crates/tinybus/src/message/mod.rs | 14 ++++++ crates/tinybus/src/router.rs | 14 ++++++ target-forged-reply-green.log | 12 ++++++ 5 files changed, 119 insertions(+), 16 deletions(-) create mode 100644 target-forged-reply-green.log diff --git a/crates/tinybus/src/broker.rs b/crates/tinybus/src/broker.rs index 740ac8ea..347a249d 100644 --- a/crates/tinybus/src/broker.rs +++ b/crates/tinybus/src/broker.rs @@ -220,17 +220,11 @@ impl Broker { // connection, which chose to participate in the exchange. let confidential_call = message.header.confidential && message.header.kind == MessageKind::MethodCall; - let target = if confidential_call { - self.router - .lock() - .expect("router lock") - .resolve_attested(&destination) - } else { - self.router - .lock() - .expect("router lock") - .resolve(&destination) - }; + let target = self.router.lock().expect("router lock").resolve_call( + &destination, + message.header.expected_responder.as_ref(), + confidential_call, + ); #[cfg(feature = "modules")] let target = target.map_err(|error| { // A refused attestation is the more specific answer and @@ -386,6 +380,13 @@ impl Broker { let (name,): (BusName,) = parse_args(member, body)?; Ok(serde_json::to_value(router.attestation_of(&name))?) } + // This method also negotiates conditional owner routing. An + // older broker fails loudly rather than ignoring the guard. + "ResolveCallTarget" => { + let (name,): (BusName,) = parse_args(member, body)?; + let owner = router.owner_of(&name).ok_or_else(|| Error::NameHasNoOwner(name))?; + Ok(serde_json::to_value(owner)?) + } "GetNameOwner" => { let (name,): (BusName,) = parse_args(member, body)?; Ok(serde_json::to_value(router.owner_of(&name))?) @@ -614,6 +615,7 @@ impl Broker { kind: MessageKind::Signal, serial: 0, reply_serial: None, + expected_responder: None, sender: Some( BusName::new(crate::BUS_NAME).expect("the bus name constant is valid"), ), diff --git a/crates/tinybus/src/connection.rs b/crates/tinybus/src/connection.rs index df3e5632..ffe104b8 100644 --- a/crates/tinybus/src/connection.rs +++ b/crates/tinybus/src/connection.rs @@ -141,6 +141,33 @@ pub const SIGNAL_BUFFER: usize = 256; /// growth in the process doing the publishing. pub const OUTBOX_CAPACITY: usize = 1024; +/// Reply identity is captured before sending, never inferred from its body. +struct PendingReply { + sender: oneshot::Sender, + expected: Option, + bus_control: bool, + confidential: bool, +} + +impl PendingReply { + fn accepts(&self, reply: &Message, brokered: bool) -> bool { + let sender = reply.header.sender.as_ref(); + let broker_sender = sender.is_some_and(|sender| sender.as_str() == crate::BUS_NAME); + let identity_matches = if self.bus_control { + // Legacy broker control replies are unstamped; peers routed by a + // real broker can never send an unstamped frame. + sender.is_none() || broker_sender + } else if let Some(expected) = &self.expected { + sender == Some(expected) || (reply.header.kind == MessageKind::Error && broker_sender) + } else { + // Raw direct transport compatibility has no authenticated peer + // identity. A proven broker delivery must not use this escape. + !brokered + }; + identity_matches && (reply.header.kind == MessageKind::Error || !self.confidential || reply.header.confidential) + } +} + struct Inner { transport: Arc, brokered_native: bool, @@ -150,7 +177,7 @@ struct Inner { /// concurrently without interleaving two frames on the wire. outbox: mpsc::Sender, serial: AtomicU64, - pending: Mutex>>, + pending: Mutex>, objects: RwLock, /// A *std* lock, not a tokio one, so a synchronous publisher can stamp /// the sender on a locally looped-back signal without an await. @@ -712,6 +739,13 @@ impl Connection { Ok(self.call_raw_message(message, timeout).await?.body) } + async fn call_bus_with_timeout(&self, member: &str, body: Value, timeout: Duration) -> Result { + self.call_raw(Message::method_call( + BusName::new(crate::BUS_NAME)?, ObjectPath::new(crate::BUS_PATH)?, + InterfaceName::new(crate::BUS_INTERFACE)?, MemberName::new(member)?, body, + ), timeout).await + } + async fn call_raw_message(&self, mut message: Message, timeout: Duration) -> Result { let serial = self.inner.serial.fetch_add(1, Ordering::Relaxed); message.header.serial = serial; @@ -737,16 +771,42 @@ impl Connection { )); } let member = message.member_or_unknown(); + let started = tokio::time::Instant::now(); + let destination = message.header.destination.clone(); + let bus_control = destination.as_ref().is_some_and(|name| name.as_str() == crate::BUS_NAME); + let expected = if bus_control { + None + } else if let Some(destination) = destination { + if destination.is_unique() { + Some(destination) + } else if self.unique_name().is_some() { + let owner: BusName = serde_json::from_value(Box::pin(self.call_bus_with_timeout( + "ResolveCallTarget", serde_json::json!([destination]), timeout, + )).await?)?; + if !owner.is_unique() { + return Err(Error::protocol("broker returned a non-unique call target")); + } + if message.header.expected_responder.as_ref().is_some_and(|expected| expected != &owner) { + return Err(Error::NameHasNoOwner(destination)); + } + message.header.expected_responder = Some(owner.clone()); + Some(owner) + } else { + None + } + } else { None }; let (tx, rx) = oneshot::channel(); - self.inner.pending.lock().await.insert(serial, tx); + self.inner.pending.lock().await.insert(serial, PendingReply { + sender: tx, expected, bus_control, confidential: message.header.confidential, + }); if let Err(e) = self.enqueue(message).await { self.inner.pending.lock().await.remove(&serial); return Err(e); } - match tokio::time::timeout(timeout, rx).await { + match tokio::time::timeout(timeout.saturating_sub(started.elapsed()), rx).await { Ok(Ok(reply)) => match reply.header.kind { MessageKind::Error => Err(reply.into_error()), _ => Ok(reply), @@ -1206,8 +1266,9 @@ async fn dispatch_loop(inner: Arc) { }; // An unmatched reply is a timed-out call arriving late. Normal, // and nothing to do but drop it. - if let Some(waiter) = inner.pending.lock().await.remove(&serial) { - let _ = waiter.send(message); + let mut pending = inner.pending.lock().await; + if pending.get(&serial).is_some_and(|waiter| waiter.accepts(&message, brokered)) { + if let Some(waiter) = pending.remove(&serial) { let _ = waiter.sender.send(message); } } } MessageKind::Signal => { diff --git a/crates/tinybus/src/message/mod.rs b/crates/tinybus/src/message/mod.rs index 67355c25..46891149 100644 --- a/crates/tinybus/src/message/mod.rs +++ b/crates/tinybus/src/message/mod.rs @@ -64,6 +64,11 @@ pub struct Header { /// headers, so this wire field alone is not authenticated authority. #[serde(default, skip_serializing_if = "Option::is_none")] pub sender: Option, + /// Conditional call routing: deliver only while this unique peer owns the + /// destination. Optional on the wire; guarded clients require the broker's + /// `ResolveCallTarget` method before relying on this restriction. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub expected_responder: Option, /// Who the message is for. `None` on signals, which are broadcast. #[serde(default, skip_serializing_if = "Option::is_none")] pub destination: Option, @@ -141,6 +146,7 @@ impl Message { serial: 0, reply_serial: None, sender: None, + expected_responder: None, destination: Some(destination), path: Some(path), interface: Some(interface), @@ -210,6 +216,7 @@ impl Message { serial: 0, reply_serial: Some(call.serial), sender: None, + expected_responder: None, // Replies are addressed back at the *unique* name the broker // stamped, not at whatever well-known name the caller happens // to own — the caller may have released it mid-call. @@ -241,6 +248,7 @@ impl Message { serial: 0, reply_serial: Some(call.serial), sender: None, + expected_responder: None, destination: call.sender.clone(), path: None, interface: None, @@ -272,6 +280,7 @@ impl Message { serial: 0, reply_serial: None, sender: None, + expected_responder: None, destination: None, path: Some(path), interface: Some(interface), @@ -326,6 +335,11 @@ impl Message { "only a method call can request a streamed reply", )); } + if let Some(expected) = &self.header.expected_responder + && (self.header.kind != MessageKind::MethodCall || !expected.is_unique()) + { + return Err(Error::protocol("expected responder must be unique and only accompany a call")); + } // Checked before the per-kind rules, and checked on ingress rather than // at delivery: a confidential signal has no destination, so there is no // one recipient to attest and fan-out is the only thing it could mean. diff --git a/crates/tinybus/src/router.rs b/crates/tinybus/src/router.rs index afebd131..ec148874 100644 --- a/crates/tinybus/src/router.rs +++ b/crates/tinybus/src/router.rs @@ -483,6 +483,20 @@ impl Router { Ok(peer.outbox.clone()) } + /// Compare a captured unique owner and select the destination under one + /// routing lock, preserving recipient attestation for confidential calls. + pub(crate) fn resolve_call( + &self, + destination: &BusName, + expected: Option<&BusName>, + confidential: bool, + ) -> Result> { + if expected.is_some_and(|expected| self.owner_of(destination).as_ref() != Some(expected)) { + return Err(Error::NameHasNoOwner(destination.clone())); + } + if confidential { self.resolve_attested(destination) } else { self.resolve(destination) } + } + /// The outbox of whoever owns `destination`. pub fn resolve(&self, destination: &BusName) -> Result> { let id = self diff --git a/target-forged-reply-green.log b/target-forged-reply-green.log new file mode 100644 index 00000000..05266073 --- /dev/null +++ b/target-forged-reply-green.log @@ -0,0 +1,12 @@ +[ci-cancel-aware] exec: cargo test -p tinybus another_broker_peer_cannot_complete + Blocking waiting for file lock on package cache + Compiling tinybus v0.1.3 (/home/enamakel/work/openhuman/worktrees/cloud-storage-7419/vendor/tinybus/crates/tinybus) + Compiling tinybus-module v0.1.3 (/home/enamakel/work/openhuman/worktrees/cloud-storage-7419/vendor/tinybus/crates/tinybus-module) + Finished `test` profile [unoptimized + debuginfo] target(s) in 16.40s + Running unittests src/lib.rs (target/debug/deps/tinybus-49c252397df0a704) + +running 1 test +test connection::reply_tests::another_broker_peer_cannot_complete_a_waiter_with_a_forged_reply ... ok + +test result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 282 filtered out; finished in 0.00s + From 31633b2dc5da23ec89a89b29e72f51f084b1edca Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 11 Oct 2026 19:21:26 +0300 Subject: [PATCH 03/14] chore: files changed crates/tinybus/src/connection_reply_tests.rs,target-reply-tests.log Auto-committed-on: dragonfly Co-authored-by: Medulla --- crates/tinybus/src/connection_reply_tests.rs | 154 +++++++++++++++++++ target-reply-tests.log | 70 +++++++++ 2 files changed, 224 insertions(+) create mode 100644 target-reply-tests.log diff --git a/crates/tinybus/src/connection_reply_tests.rs b/crates/tinybus/src/connection_reply_tests.rs index 72dd244e..f543c02a 100644 --- a/crates/tinybus/src/connection_reply_tests.rs +++ b/crates/tinybus/src/connection_reply_tests.rs @@ -1,3 +1,5 @@ +//! Reply authorization regressions over broker-stamped memory deliveries. + use super::*; use crate::broker::Broker; use crate::transport::memory::MemoryBus; @@ -41,3 +43,155 @@ async fn another_broker_peer_cannot_complete_a_waiter_with_a_forged_reply() { responder.send(genuine).await.unwrap(); assert_eq!(pending.await.unwrap().unwrap(), serde_json::json!("genuine")); } + + +fn fixture_call(destination: BusName) -> Message { + Message::method_call(destination, ObjectPath::new("/fixture").unwrap(), + InterfaceName::new("ai.tinyhumans.Fixture").unwrap(), + MemberName::new("Await").unwrap(), serde_json::json!([])) +} + +async fn raw_control(peer: &Arc, serial: u64, member: &str, body: Value) -> Message { + let mut call = Message::method_call(BusName::new(crate::BUS_NAME).unwrap(), + ObjectPath::new(crate::BUS_PATH).unwrap(), + InterfaceName::new(crate::BUS_INTERFACE).unwrap(), + MemberName::new(member).unwrap(), body); + call.header.serial = serial; + peer.send(call).await.unwrap(); + loop { + let reply = peer.recv().await.unwrap().unwrap(); + if reply.header.reply_serial == Some(serial) { return reply; } + } +} + +async fn reject_forged_replies(named: bool, confidential: bool, genuine_error: bool) { + let bus = MemoryBus::new(); + let broker = Broker::new(); + broker.clone().spawn(bus.clone()); + let responder = bus.connect().await.unwrap(); + let hello = raw_control(&responder, 1, "Hello", serde_json::json!([])).await; + let responder_name = BusName::new(hello.body.as_str().unwrap()).unwrap(); + let alias = BusName::new("ai.tinyhumans.Fixture").unwrap(); + if named { + let reply = raw_control(&responder, 2, "RequestName", serde_json::json!([alias])).await; + assert_eq!(reply.header.kind, MessageKind::MethodReturn); + } + #[cfg(feature = "modules")] + if confidential { + broker.attest_module(&responder_name, crate::attest::Attestation { + name: alias.clone(), sha256: "a".repeat(64), + }); + } + let requester = Connection::connect(bus.connect().await.unwrap()).await.unwrap(); + let forger = Connection::connect(bus.connect().await.unwrap()).await.unwrap(); + let mut barriers = requester.add_match(MatchRule::new().signals()).await.unwrap(); + let mut request = fixture_call(if named { alias } else { responder_name.clone() }); + request.header.confidential = confidential; + let pending = tokio::spawn({ let requester = requester.clone(); async move { + requester.call_raw(request, DEFAULT_TIMEOUT).await + }}); + let delivered = responder.recv().await.unwrap().unwrap(); + if named { assert_eq!(delivered.header.expected_responder, Some(responder_name.clone())); } + for error in [false, true] { + for private in [false, true] { + let mut forged = if error { + Message::error_reply(&delivered.header, &Error::ConnectionClosed) + } else { + Message::method_return(&delivered.header, serde_json::json!("forged")) + }; + forged.header.confidential = private; + forged.header.sensitive = private; + forged.header.sender = Some(BusName::new(crate::BUS_NAME).unwrap()); + forger.send(forged).await.unwrap(); + } + } + forger.emit(ObjectPath::new("/fixture").unwrap(), + InterfaceName::new("ai.tinyhumans.Fixture").unwrap(), + MemberName::new("Barrier").unwrap(), ()).await.unwrap(); + let barrier = tokio::time::timeout(DEFAULT_TIMEOUT, barriers.recv()).await.unwrap().unwrap(); + assert_eq!(barrier.header.sender, forger.unique_name()); + let mut genuine = if genuine_error { + Message::error_reply(&delivered.header, &Error::UnknownMethod { + interface: "ai.tinyhumans.Fixture".into(), member: "Await".into(), + }) + } else { + Message::method_return(&delivered.header, serde_json::json!("genuine")) + }; + genuine.header.serial = 3; + responder.send(genuine).await.unwrap(); + let result = pending.await.unwrap(); + if genuine_error { + let error = result.unwrap_err(); + assert!(matches!(error, Error::MethodFailed { ref name, .. } + if name == Error::UnknownMethod { interface: "ai.tinyhumans.Fixture".into(), member: "Await".into() }.wire_name())); + } else { assert_eq!(result.unwrap(), serde_json::json!("genuine")); } +} + +#[tokio::test] +async fn named_reply_rejects_forged_success_and_errors_with_any_private_flags() { + reject_forged_replies(true, false, false).await; +} + +#[tokio::test] +async fn unique_reply_preserves_genuine_error_after_rejecting_forged_errors() { + reject_forged_replies(false, false, true).await; +} + +#[cfg(feature = "modules")] +#[tokio::test] +async fn confidential_named_reply_binds_the_attested_owner() { + reject_forged_replies(true, true, false).await; +} + +#[cfg(feature = "modules")] +#[tokio::test] +async fn confidential_unique_reply_preserves_genuine_sanitized_error() { + reject_forged_replies(false, true, true).await; +} + +#[tokio::test] +async fn captured_alias_owner_is_checked_atomically_before_dispatch() { + let bus = MemoryBus::new(); + Broker::new().spawn(bus.clone()); + let original = Connection::connect(bus.connect().await.unwrap()).await.unwrap(); + let replacement = Connection::connect(bus.connect().await.unwrap()).await.unwrap(); + let requester = Connection::connect(bus.connect().await.unwrap()).await.unwrap(); + let alias = BusName::new("ai.tinyhumans.Fixture").unwrap(); + original.request_name(alias.clone()).await.unwrap(); + let captured: BusName = serde_json::from_value(requester.call_bus( + "ResolveCallTarget", serde_json::json!([alias])).await.unwrap()).unwrap(); + original.release_name(alias.clone()).await.unwrap(); + replacement.request_name(alias.clone()).await.unwrap(); + let mut call = fixture_call(alias.clone()); + call.header.expected_responder = Some(captured); + // Send through the low-level port, bypassing the SDK's earlier check, to + // pin the broker's routing check independently of requester validation. + call.header.serial = 900; + let mut replies = requester.add_match(MatchRule::new().signals()).await.unwrap(); + requester.send(call).await.unwrap(); + requester.emit(ObjectPath::new("/fixture").unwrap(), + InterfaceName::new("ai.tinyhumans.Fixture").unwrap(), + MemberName::new("Barrier").unwrap(), ()).await.unwrap(); + tokio::time::timeout(DEFAULT_TIMEOUT, replies.recv()).await.unwrap().unwrap(); + // Exercise the SDK pre-send check as well; the replacement must never be + // invoked with a stale binding. + let mut stale = fixture_call(alias.clone()); + stale.header.expected_responder = Some(original.unique_name().unwrap()); + assert!(matches!(requester.call_raw(stale, DEFAULT_TIMEOUT).await, + Err(Error::NameHasNoOwner(name)) if name == alias)); +} + +#[tokio::test] +async fn raw_direct_reply_compatibility_does_not_claim_broker_identity() { + let (a, b) = crate::transport::memory::pair(16); + let requester = Connection::attach(Arc::new(a)); + let responder = Arc::new(b); + let pending = tokio::spawn({ let requester = requester.clone(); async move { + requester.call_raw(fixture_call(BusName::new("ai.tinyhumans.Fixture").unwrap()), DEFAULT_TIMEOUT).await + }}); + let delivered = responder.recv().await.unwrap().unwrap(); + let mut response = Message::method_return(&delivered.header, serde_json::json!("raw")); + response.header.serial = 1; + responder.send(response).await.unwrap(); + assert_eq!(pending.await.unwrap().unwrap(), serde_json::json!("raw")); +} diff --git a/target-reply-tests.log b/target-reply-tests.log new file mode 100644 index 00000000..424905cd --- /dev/null +++ b/target-reply-tests.log @@ -0,0 +1,70 @@ +[ci-cancel-aware] exec: cargo test -p tinybus --all-features reply_tests + Compiling find-msvc-tools v0.1.10 + Compiling shlex v2.0.1 + Compiling serde_core v1.0.229 + Compiling crc32fast v1.5.0 + Compiling simd-adler32 v0.3.10 + Compiling log v0.4.33 + Compiling cc v1.4.2 + Compiling zeroize v1.9.0 + Compiling rustls-pki-types v1.15.1 + Compiling ring v0.17.14 + Compiling getrandom v0.2.17 + Compiling adler2 v2.0.1 + Compiling untrusted v0.9.0 + Compiling httparse v1.10.1 + Compiling utf8parse v0.2.2 + Compiling miniz_oxide v0.8.9 + Compiling anstyle-parse v1.0.0 + Compiling regex-syntax v0.8.11 + Compiling colorchoice v1.0.5 + Compiling anstyle v1.0.14 + Compiling rustls v0.23.43 + Compiling is_terminal_polyfill v1.70.2 + Compiling anstyle-query v1.1.5 + Compiling flate2 v1.1.9 + Compiling regex-automata v0.4.18 + Compiling anstream v1.0.0 + Compiling http v1.5.0 + Compiling errno v0.3.14 + Compiling memchr v2.8.3 + Compiling bumpalo v3.20.3 + Compiling rustls-webpki v0.103.13 + Compiling hashbrown v0.17.1 + Compiling equivalent v1.0.2 + Compiling zip v2.4.2 + Compiling clap_lex v1.1.0 + Compiling base64 v0.23.1 + Compiling subtle v2.6.1 + Compiling heck v0.5.0 + Compiling lazy_static v1.5.0 + Compiling strsim v0.11.1 + Compiling winnow v1.0.4 + Compiling clap_derive v4.6.4 + Compiling sharded-slab v0.1.7 + Compiling clap_builder v4.6.6 + Compiling toml_parser v1.1.3+spec-1.1.0 + Compiling ureq-proto v0.6.1 + Compiling indexmap v2.14.0 + Compiling matchers v0.2.0 + Compiling zopfli v0.8.3 + Compiling signal-hook-registry v1.4.8 + Compiling serde_spanned v1.1.1 + Compiling toml_datetime v1.1.1+spec-1.1.0 + Compiling xattr v1.6.1 + Compiling webpki-roots v1.0.9 + Compiling displaydoc v0.2.7 + Compiling filetime v0.2.29 + Compiling thread_local v1.1.10 + Compiling tinybus v0.1.3 (/home/enamakel/work/openhuman/worktrees/cloud-storage-7419/vendor/tinybus/crates/tinybus) + Compiling toml_writer v1.1.2+spec-1.1.0 + Compiling percent-encoding v2.3.2 + Compiling utf8-zero v0.8.1 + Compiling toml v1.1.6+spec-1.1.0 + Compiling tracing-subscriber v0.3.23 + Compiling tar v0.4.46 + Compiling ureq v3.4.0 + Compiling tokio v1.53.1 + Compiling clap v4.6.6 + Compiling serde v1.0.229 + Compiling serde_json v1.0.151 From 15d32c5da7f5b8594b7211988b4d65fbc6ec7eb7 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 11 Oct 2026 19:21:52 +0300 Subject: [PATCH 04/14] chore: files changed crates/tinybus/src/connection_reply_tests.rs,target-reply-tests.log Auto-committed-on: dragonfly Co-authored-by: Medulla --- crates/tinybus/src/connection_reply_tests.rs | 48 +++++---- target-reply-tests.log | 100 +++++++++++++++++++ 2 files changed, 129 insertions(+), 19 deletions(-) diff --git a/crates/tinybus/src/connection_reply_tests.rs b/crates/tinybus/src/connection_reply_tests.rs index f543c02a..d812230d 100644 --- a/crates/tinybus/src/connection_reply_tests.rs +++ b/crates/tinybus/src/connection_reply_tests.rs @@ -154,36 +154,46 @@ async fn captured_alias_owner_is_checked_atomically_before_dispatch() { let bus = MemoryBus::new(); Broker::new().spawn(bus.clone()); let original = Connection::connect(bus.connect().await.unwrap()).await.unwrap(); - let replacement = Connection::connect(bus.connect().await.unwrap()).await.unwrap(); - let requester = Connection::connect(bus.connect().await.unwrap()).await.unwrap(); + let replacement = bus.connect().await.unwrap(); + let requester = bus.connect().await.unwrap(); + raw_control(&replacement, 1, "Hello", serde_json::json!([])).await; + raw_control(&requester, 1, "Hello", serde_json::json!([])).await; let alias = BusName::new("ai.tinyhumans.Fixture").unwrap(); original.request_name(alias.clone()).await.unwrap(); - let captured: BusName = serde_json::from_value(requester.call_bus( - "ResolveCallTarget", serde_json::json!([alias])).await.unwrap()).unwrap(); + let captured_reply = raw_control(&requester, 2, "ResolveCallTarget", serde_json::json!([alias])).await; + let captured: BusName = serde_json::from_value(captured_reply.body).unwrap(); original.release_name(alias.clone()).await.unwrap(); - replacement.request_name(alias.clone()).await.unwrap(); + raw_control(&replacement, 2, "RequestName", serde_json::json!([alias])).await; let mut call = fixture_call(alias.clone()); - call.header.expected_responder = Some(captured); - // Send through the low-level port, bypassing the SDK's earlier check, to - // pin the broker's routing check independently of requester validation. - call.header.serial = 900; - let mut replies = requester.add_match(MatchRule::new().signals()).await.unwrap(); + call.header.expected_responder = Some(captured.clone()); + call.header.serial = 3; requester.send(call).await.unwrap(); - requester.emit(ObjectPath::new("/fixture").unwrap(), - InterfaceName::new("ai.tinyhumans.Fixture").unwrap(), - MemberName::new("Barrier").unwrap(), ()).await.unwrap(); - tokio::time::timeout(DEFAULT_TIMEOUT, replies.recv()).await.unwrap().unwrap(); - // Exercise the SDK pre-send check as well; the replacement must never be - // invoked with a stale binding. + let refusal = requester.recv().await.unwrap().unwrap(); + assert_eq!(refusal.header.reply_serial, Some(3)); + assert_eq!(refusal.header.kind, MessageKind::Error); + assert_eq!(refusal.header.sender.as_ref().unwrap().as_str(), crate::BUS_NAME); + assert_eq!(refusal.header.error_name.as_deref(), Some(Error::NAME_HAS_NO_OWNER)); + // An ordered broker control reply to the replacement is a barrier after + // the refused call: an incorrectly routed call would arrive before it. + let mut ping = fixture_call(BusName::new(crate::BUS_NAME).unwrap()); + ping.header.path = Some(ObjectPath::new(crate::BUS_PATH).unwrap()); + ping.header.interface = Some(InterfaceName::new(crate::BUS_INTERFACE).unwrap()); + ping.header.member = Some(MemberName::new("Ping").unwrap()); + ping.header.serial = 3; + replacement.send(ping).await.unwrap(); + let first = replacement.recv().await.unwrap().unwrap(); + assert_eq!(first.header.kind, MessageKind::MethodReturn); + assert_eq!(first.header.reply_serial, Some(3)); + let sdk = Connection::connect(bus.connect().await.unwrap()).await.unwrap(); let mut stale = fixture_call(alias.clone()); - stale.header.expected_responder = Some(original.unique_name().unwrap()); - assert!(matches!(requester.call_raw(stale, DEFAULT_TIMEOUT).await, + stale.header.expected_responder = Some(captured); + assert!(matches!(sdk.call_raw(stale, DEFAULT_TIMEOUT).await, Err(Error::NameHasNoOwner(name)) if name == alias)); } #[tokio::test] async fn raw_direct_reply_compatibility_does_not_claim_broker_identity() { - let (a, b) = crate::transport::memory::pair(16); + let (a, b) = crate::transport::memory::MemoryTransport::pair(); let requester = Connection::attach(Arc::new(a)); let responder = Arc::new(b); let pending = tokio::spawn({ let requester = requester.clone(); async move { diff --git a/target-reply-tests.log b/target-reply-tests.log index 424905cd..a101ddc9 100644 --- a/target-reply-tests.log +++ b/target-reply-tests.log @@ -68,3 +68,103 @@ Compiling clap v4.6.6 Compiling serde v1.0.229 Compiling serde_json v1.0.151 + Compiling tinybus-module v0.1.3 (/home/enamakel/work/openhuman/worktrees/cloud-storage-7419/vendor/tinybus/crates/tinybus-module) +error[E0425]: cannot find function `pair` in module `crate::transport::memory` + --> crates/tinybus/src/connection_reply_tests.rs:186:44 + | +186 | let (a, b) = crate::transport::memory::pair(16); + | ^^^^ not found in `crate::transport::memory` + | +note: function `crate::connection::tests::pair` exists but is inaccessible + --> crates/tinybus/src/connection_tests.rs:58:1 + | + 58 | async fn pair() -> (Connection, Connection) { + | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ not accessible + +error[E0308]: mismatched types + --> crates/tinybus/src/connection_reply_tests.rs:72:29 + | +72 | let hello = raw_control(&responder, 1, "Hello", serde_json::json!([])).await; + | ----------- ^^^^^^^^^^ expected `&Arc`, found `&Box` + | | + | arguments to this function are incorrect + | + = note: expected reference `&std::sync::Arc<(dyn ports::transport::Transport + 'static)>` + found reference `&Box` +note: function defined here + --> crates/tinybus/src/connection_reply_tests.rs:54:10 + | +54 | async fn raw_control(peer: &Arc, serial: u64, member: &str, body: Value) -> Message { + | ^^^^^^^^^^^ ------------------------- + +error[E0308]: mismatched types + --> crates/tinybus/src/connection_reply_tests.rs:76:33 + | +76 | let reply = raw_control(&responder, 2, "RequestName", serde_json::json!([alias])).await; + | ----------- ^^^^^^^^^^ expected `&Arc`, found `&Box` + | | + | arguments to this function are incorrect + | + = note: expected reference `&std::sync::Arc<(dyn ports::transport::Transport + 'static)>` + found reference `&Box` +note: function defined here + --> crates/tinybus/src/connection_reply_tests.rs:54:10 + | +54 | async fn raw_control(peer: &Arc, serial: u64, member: &str, body: Value) -> Message { + | ^^^^^^^^^^^ ------------------------- + +error[E0277]: the trait bound `name::InterfaceName: From<&str>` is not satisfied + --> crates/tinybus/src/connection_reply_tests.rs:115:48 + | +115 | interface: "ai.tinyhumans.Fixture".into(), member: "Await".into(), + | ^^^^ unsatisfied trait bound + | +help: the trait `From<&str>` is not implemented for `name::InterfaceName` + --> crates/tinybus/src/name.rs:222:1 + | +222 | pub struct InterfaceName(String); + | ^^^^^^^^^^^^^^^^^^^^^^^^ + = note: required for `&str` to implement `Into` + +error[E0277]: the trait bound `name::MemberName: From<&str>` is not satisfied + --> crates/tinybus/src/connection_reply_tests.rs:115:72 + | +115 | interface: "ai.tinyhumans.Fixture".into(), member: "Await".into(), + | ^^^^ unsatisfied trait bound + | +help: the trait `From<&str>` is not implemented for `name::MemberName` + --> crates/tinybus/src/name.rs:245:1 + | +245 | pub struct MemberName(String); + | ^^^^^^^^^^^^^^^^^^^^^ + = note: required for `&str` to implement `Into` + +error[E0277]: the trait bound `name::InterfaceName: From<&str>` is not satisfied + --> crates/tinybus/src/connection_reply_tests.rs:126:82 + | +126 | if name == Error::UnknownMethod { interface: "ai.tinyhumans.Fixture".into(), member: "Await".into() }.wire_name())); + | ^^^^ unsatisfied trait bound + | +help: the trait `From<&str>` is not implemented for `name::InterfaceName` + --> crates/tinybus/src/name.rs:222:1 + | +222 | pub struct InterfaceName(String); + | ^^^^^^^^^^^^^^^^^^^^^^^^ + = note: required for `&str` to implement `Into` + +error[E0277]: the trait bound `name::MemberName: From<&str>` is not satisfied + --> crates/tinybus/src/connection_reply_tests.rs:126:106 + | +126 | if name == Error::UnknownMethod { interface: "ai.tinyhumans.Fixture".into(), member: "Await".into() }.wire_name())); + | ^^^^ unsatisfied trait bound + | +help: the trait `From<&str>` is not implemented for `name::MemberName` + --> crates/tinybus/src/name.rs:245:1 + | +245 | pub struct MemberName(String); + | ^^^^^^^^^^^^^^^^^^^^^ + = note: required for `&str` to implement `Into` + +Some errors have detailed explanations: E0277, E0308, E0425. +For more information about an error, try `rustc --explain E0277`. +error: could not compile `tinybus` (lib test) due to 7 previous errors From 139453b9e0e0e2ba693603b4d6b00b999bffc4b0 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 11 Oct 2026 19:22:48 +0300 Subject: [PATCH 05/14] test(tinybus): fix reply test compilation errors Update the reply tests to take a plain transport reference instead of an Arc, and construct InterfaceName and MemberName via their constructors rather than From<&str>, which is not implemented. The captured test log is refreshed to reflect the remaining compile error. Auto-committed-on: dragonfly Co-authored-by: Medulla --- crates/tinybus/src/connection_reply_tests.rs | 18 +- target-reply-tests.log | 173 +------------------ 2 files changed, 18 insertions(+), 173 deletions(-) diff --git a/crates/tinybus/src/connection_reply_tests.rs b/crates/tinybus/src/connection_reply_tests.rs index d812230d..20e8b1dc 100644 --- a/crates/tinybus/src/connection_reply_tests.rs +++ b/crates/tinybus/src/connection_reply_tests.rs @@ -51,7 +51,7 @@ fn fixture_call(destination: BusName) -> Message { MemberName::new("Await").unwrap(), serde_json::json!([])) } -async fn raw_control(peer: &Arc, serial: u64, member: &str, body: Value) -> Message { +async fn raw_control(peer: &dyn Transport, serial: u64, member: &str, body: Value) -> Message { let mut call = Message::method_call(BusName::new(crate::BUS_NAME).unwrap(), ObjectPath::new(crate::BUS_PATH).unwrap(), InterfaceName::new(crate::BUS_INTERFACE).unwrap(), @@ -69,11 +69,11 @@ async fn reject_forged_replies(named: bool, confidential: bool, genuine_error: b let broker = Broker::new(); broker.clone().spawn(bus.clone()); let responder = bus.connect().await.unwrap(); - let hello = raw_control(&responder, 1, "Hello", serde_json::json!([])).await; + let hello = raw_control(responder.as_ref(), 1, "Hello", serde_json::json!([])).await; let responder_name = BusName::new(hello.body.as_str().unwrap()).unwrap(); let alias = BusName::new("ai.tinyhumans.Fixture").unwrap(); if named { - let reply = raw_control(&responder, 2, "RequestName", serde_json::json!([alias])).await; + let reply = raw_control(responder.as_ref(), 2, "RequestName", serde_json::json!([alias])).await; assert_eq!(reply.header.kind, MessageKind::MethodReturn); } #[cfg(feature = "modules")] @@ -112,7 +112,7 @@ async fn reject_forged_replies(named: bool, confidential: bool, genuine_error: b assert_eq!(barrier.header.sender, forger.unique_name()); let mut genuine = if genuine_error { Message::error_reply(&delivered.header, &Error::UnknownMethod { - interface: "ai.tinyhumans.Fixture".into(), member: "Await".into(), + interface: InterfaceName::new("ai.tinyhumans.Fixture").unwrap(), member: MemberName::new("Await").unwrap(), }) } else { Message::method_return(&delivered.header, serde_json::json!("genuine")) @@ -123,7 +123,7 @@ async fn reject_forged_replies(named: bool, confidential: bool, genuine_error: b if genuine_error { let error = result.unwrap_err(); assert!(matches!(error, Error::MethodFailed { ref name, .. } - if name == Error::UnknownMethod { interface: "ai.tinyhumans.Fixture".into(), member: "Await".into() }.wire_name())); + if name == Error::UnknownMethod { interface: InterfaceName::new("ai.tinyhumans.Fixture").unwrap(), member: MemberName::new("Await").unwrap() }.wire_name())); } else { assert_eq!(result.unwrap(), serde_json::json!("genuine")); } } @@ -156,14 +156,14 @@ async fn captured_alias_owner_is_checked_atomically_before_dispatch() { let original = Connection::connect(bus.connect().await.unwrap()).await.unwrap(); let replacement = bus.connect().await.unwrap(); let requester = bus.connect().await.unwrap(); - raw_control(&replacement, 1, "Hello", serde_json::json!([])).await; - raw_control(&requester, 1, "Hello", serde_json::json!([])).await; + raw_control(replacement.as_ref(), 1, "Hello", serde_json::json!([])).await; + raw_control(requester.as_ref(), 1, "Hello", serde_json::json!([])).await; let alias = BusName::new("ai.tinyhumans.Fixture").unwrap(); original.request_name(alias.clone()).await.unwrap(); - let captured_reply = raw_control(&requester, 2, "ResolveCallTarget", serde_json::json!([alias])).await; + let captured_reply = raw_control(requester.as_ref(), 2, "ResolveCallTarget", serde_json::json!([alias])).await; let captured: BusName = serde_json::from_value(captured_reply.body).unwrap(); original.release_name(alias.clone()).await.unwrap(); - raw_control(&replacement, 2, "RequestName", serde_json::json!([alias])).await; + raw_control(replacement.as_ref(), 2, "RequestName", serde_json::json!([alias])).await; let mut call = fixture_call(alias.clone()); call.header.expected_responder = Some(captured.clone()); call.header.serial = 3; diff --git a/target-reply-tests.log b/target-reply-tests.log index a101ddc9..4e0fb047 100644 --- a/target-reply-tests.log +++ b/target-reply-tests.log @@ -1,170 +1,15 @@ [ci-cancel-aware] exec: cargo test -p tinybus --all-features reply_tests - Compiling find-msvc-tools v0.1.10 - Compiling shlex v2.0.1 - Compiling serde_core v1.0.229 - Compiling crc32fast v1.5.0 - Compiling simd-adler32 v0.3.10 - Compiling log v0.4.33 - Compiling cc v1.4.2 - Compiling zeroize v1.9.0 - Compiling rustls-pki-types v1.15.1 - Compiling ring v0.17.14 - Compiling getrandom v0.2.17 - Compiling adler2 v2.0.1 - Compiling untrusted v0.9.0 - Compiling httparse v1.10.1 - Compiling utf8parse v0.2.2 - Compiling miniz_oxide v0.8.9 - Compiling anstyle-parse v1.0.0 - Compiling regex-syntax v0.8.11 - Compiling colorchoice v1.0.5 - Compiling anstyle v1.0.14 - Compiling rustls v0.23.43 - Compiling is_terminal_polyfill v1.70.2 - Compiling anstyle-query v1.1.5 - Compiling flate2 v1.1.9 - Compiling regex-automata v0.4.18 - Compiling anstream v1.0.0 - Compiling http v1.5.0 - Compiling errno v0.3.14 - Compiling memchr v2.8.3 - Compiling bumpalo v3.20.3 - Compiling rustls-webpki v0.103.13 - Compiling hashbrown v0.17.1 - Compiling equivalent v1.0.2 - Compiling zip v2.4.2 - Compiling clap_lex v1.1.0 - Compiling base64 v0.23.1 - Compiling subtle v2.6.1 - Compiling heck v0.5.0 - Compiling lazy_static v1.5.0 - Compiling strsim v0.11.1 - Compiling winnow v1.0.4 - Compiling clap_derive v4.6.4 - Compiling sharded-slab v0.1.7 - Compiling clap_builder v4.6.6 - Compiling toml_parser v1.1.3+spec-1.1.0 - Compiling ureq-proto v0.6.1 - Compiling indexmap v2.14.0 - Compiling matchers v0.2.0 - Compiling zopfli v0.8.3 - Compiling signal-hook-registry v1.4.8 - Compiling serde_spanned v1.1.1 - Compiling toml_datetime v1.1.1+spec-1.1.0 - Compiling xattr v1.6.1 - Compiling webpki-roots v1.0.9 - Compiling displaydoc v0.2.7 - Compiling filetime v0.2.29 - Compiling thread_local v1.1.10 Compiling tinybus v0.1.3 (/home/enamakel/work/openhuman/worktrees/cloud-storage-7419/vendor/tinybus/crates/tinybus) - Compiling toml_writer v1.1.2+spec-1.1.0 - Compiling percent-encoding v2.3.2 - Compiling utf8-zero v0.8.1 - Compiling toml v1.1.6+spec-1.1.0 - Compiling tracing-subscriber v0.3.23 - Compiling tar v0.4.46 - Compiling ureq v3.4.0 - Compiling tokio v1.53.1 - Compiling clap v4.6.6 - Compiling serde v1.0.229 - Compiling serde_json v1.0.151 - Compiling tinybus-module v0.1.3 (/home/enamakel/work/openhuman/worktrees/cloud-storage-7419/vendor/tinybus/crates/tinybus-module) -error[E0425]: cannot find function `pair` in module `crate::transport::memory` - --> crates/tinybus/src/connection_reply_tests.rs:186:44 +error[E0599]: no variant, associated function, or constant named `NAME_HAS_NO_OWNER` found for enum `error::Error` in the current scope + --> crates/tinybus/src/connection_reply_tests.rs:175:66 | -186 | let (a, b) = crate::transport::memory::pair(16); - | ^^^^ not found in `crate::transport::memory` +175 | assert_eq!(refusal.header.error_name.as_deref(), Some(Error::NAME_HAS_NO_OWNER)); + | ^^^^^^^^^^^^^^^^^ variant, associated function, or constant not found in `error::Error` | -note: function `crate::connection::tests::pair` exists but is inaccessible - --> crates/tinybus/src/connection_tests.rs:58:1 + ::: crates/tinybus/src/error.rs:22:1 | - 58 | async fn pair() -> (Connection, Connection) { - | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ not accessible + 22 | pub enum Error { + | -------------- variant, associated function, or constant `NAME_HAS_NO_OWNER` not found for this enum -error[E0308]: mismatched types - --> crates/tinybus/src/connection_reply_tests.rs:72:29 - | -72 | let hello = raw_control(&responder, 1, "Hello", serde_json::json!([])).await; - | ----------- ^^^^^^^^^^ expected `&Arc`, found `&Box` - | | - | arguments to this function are incorrect - | - = note: expected reference `&std::sync::Arc<(dyn ports::transport::Transport + 'static)>` - found reference `&Box` -note: function defined here - --> crates/tinybus/src/connection_reply_tests.rs:54:10 - | -54 | async fn raw_control(peer: &Arc, serial: u64, member: &str, body: Value) -> Message { - | ^^^^^^^^^^^ ------------------------- - -error[E0308]: mismatched types - --> crates/tinybus/src/connection_reply_tests.rs:76:33 - | -76 | let reply = raw_control(&responder, 2, "RequestName", serde_json::json!([alias])).await; - | ----------- ^^^^^^^^^^ expected `&Arc`, found `&Box` - | | - | arguments to this function are incorrect - | - = note: expected reference `&std::sync::Arc<(dyn ports::transport::Transport + 'static)>` - found reference `&Box` -note: function defined here - --> crates/tinybus/src/connection_reply_tests.rs:54:10 - | -54 | async fn raw_control(peer: &Arc, serial: u64, member: &str, body: Value) -> Message { - | ^^^^^^^^^^^ ------------------------- - -error[E0277]: the trait bound `name::InterfaceName: From<&str>` is not satisfied - --> crates/tinybus/src/connection_reply_tests.rs:115:48 - | -115 | interface: "ai.tinyhumans.Fixture".into(), member: "Await".into(), - | ^^^^ unsatisfied trait bound - | -help: the trait `From<&str>` is not implemented for `name::InterfaceName` - --> crates/tinybus/src/name.rs:222:1 - | -222 | pub struct InterfaceName(String); - | ^^^^^^^^^^^^^^^^^^^^^^^^ - = note: required for `&str` to implement `Into` - -error[E0277]: the trait bound `name::MemberName: From<&str>` is not satisfied - --> crates/tinybus/src/connection_reply_tests.rs:115:72 - | -115 | interface: "ai.tinyhumans.Fixture".into(), member: "Await".into(), - | ^^^^ unsatisfied trait bound - | -help: the trait `From<&str>` is not implemented for `name::MemberName` - --> crates/tinybus/src/name.rs:245:1 - | -245 | pub struct MemberName(String); - | ^^^^^^^^^^^^^^^^^^^^^ - = note: required for `&str` to implement `Into` - -error[E0277]: the trait bound `name::InterfaceName: From<&str>` is not satisfied - --> crates/tinybus/src/connection_reply_tests.rs:126:82 - | -126 | if name == Error::UnknownMethod { interface: "ai.tinyhumans.Fixture".into(), member: "Await".into() }.wire_name())); - | ^^^^ unsatisfied trait bound - | -help: the trait `From<&str>` is not implemented for `name::InterfaceName` - --> crates/tinybus/src/name.rs:222:1 - | -222 | pub struct InterfaceName(String); - | ^^^^^^^^^^^^^^^^^^^^^^^^ - = note: required for `&str` to implement `Into` - -error[E0277]: the trait bound `name::MemberName: From<&str>` is not satisfied - --> crates/tinybus/src/connection_reply_tests.rs:126:106 - | -126 | if name == Error::UnknownMethod { interface: "ai.tinyhumans.Fixture".into(), member: "Await".into() }.wire_name())); - | ^^^^ unsatisfied trait bound - | -help: the trait `From<&str>` is not implemented for `name::MemberName` - --> crates/tinybus/src/name.rs:245:1 - | -245 | pub struct MemberName(String); - | ^^^^^^^^^^^^^^^^^^^^^ - = note: required for `&str` to implement `Into` - -Some errors have detailed explanations: E0277, E0308, E0425. -For more information about an error, try `rustc --explain E0277`. -error: could not compile `tinybus` (lib test) due to 7 previous errors +For more information about this error, try `rustc --explain E0599`. +error: could not compile `tinybus` (lib test) due to 1 previous error From 6b8ab7309cbe21e7a4ca410bb3a702141c2d557b Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 11 Oct 2026 19:23:40 +0300 Subject: [PATCH 06/14] test(connection): cover reply authority and owner guard negotiation Expand the reply tests to exercise forged replies across named, confidential, and error cases, and add unit coverage for the pending-reply acceptance rules and the conditional owner header. The captured-alias test now compares against the wire name of the actual error variant, and the log records a clean compile. Auto-committed-on: dragonfly Co-authored-by: Medulla --- crates/tinybus/src/broker.rs | 4 +- crates/tinybus/src/connection.rs | 74 ++++- crates/tinybus/src/connection_reply_tests.rs | 332 ++++++++++++++++--- crates/tinybus/src/message/mod.rs | 4 +- crates/tinybus/src/router.rs | 6 +- target-reply-tests.log | 14 +- 6 files changed, 355 insertions(+), 79 deletions(-) diff --git a/crates/tinybus/src/broker.rs b/crates/tinybus/src/broker.rs index 347a249d..dd3fe0e9 100644 --- a/crates/tinybus/src/broker.rs +++ b/crates/tinybus/src/broker.rs @@ -384,7 +384,9 @@ impl Broker { // older broker fails loudly rather than ignoring the guard. "ResolveCallTarget" => { let (name,): (BusName,) = parse_args(member, body)?; - let owner = router.owner_of(&name).ok_or_else(|| Error::NameHasNoOwner(name))?; + let owner = router + .owner_of(&name) + .ok_or_else(|| Error::NameHasNoOwner(name))?; Ok(serde_json::to_value(owner)?) } "GetNameOwner" => { diff --git a/crates/tinybus/src/connection.rs b/crates/tinybus/src/connection.rs index ffe104b8..496c8dbd 100644 --- a/crates/tinybus/src/connection.rs +++ b/crates/tinybus/src/connection.rs @@ -164,7 +164,10 @@ impl PendingReply { // identity. A proven broker delivery must not use this escape. !brokered }; - identity_matches && (reply.header.kind == MessageKind::Error || !self.confidential || reply.header.confidential) + identity_matches + && (reply.header.kind == MessageKind::Error + || !self.confidential + || reply.header.confidential) } } @@ -739,11 +742,23 @@ impl Connection { Ok(self.call_raw_message(message, timeout).await?.body) } - async fn call_bus_with_timeout(&self, member: &str, body: Value, timeout: Duration) -> Result { - self.call_raw(Message::method_call( - BusName::new(crate::BUS_NAME)?, ObjectPath::new(crate::BUS_PATH)?, - InterfaceName::new(crate::BUS_INTERFACE)?, MemberName::new(member)?, body, - ), timeout).await + async fn call_bus_with_timeout( + &self, + member: &str, + body: Value, + timeout: Duration, + ) -> Result { + self.call_raw( + Message::method_call( + BusName::new(crate::BUS_NAME)?, + ObjectPath::new(crate::BUS_PATH)?, + InterfaceName::new(crate::BUS_INTERFACE)?, + MemberName::new(member)?, + body, + ), + timeout, + ) + .await } async fn call_raw_message(&self, mut message: Message, timeout: Duration) -> Result { @@ -773,20 +788,32 @@ impl Connection { let member = message.member_or_unknown(); let started = tokio::time::Instant::now(); let destination = message.header.destination.clone(); - let bus_control = destination.as_ref().is_some_and(|name| name.as_str() == crate::BUS_NAME); + let bus_control = destination + .as_ref() + .is_some_and(|name| name.as_str() == crate::BUS_NAME); let expected = if bus_control { None } else if let Some(destination) = destination { if destination.is_unique() { Some(destination) } else if self.unique_name().is_some() { - let owner: BusName = serde_json::from_value(Box::pin(self.call_bus_with_timeout( - "ResolveCallTarget", serde_json::json!([destination]), timeout, - )).await?)?; + let owner: BusName = serde_json::from_value( + Box::pin(self.call_bus_with_timeout( + "ResolveCallTarget", + serde_json::json!([destination]), + timeout, + )) + .await?, + )?; if !owner.is_unique() { return Err(Error::protocol("broker returned a non-unique call target")); } - if message.header.expected_responder.as_ref().is_some_and(|expected| expected != &owner) { + if message + .header + .expected_responder + .as_ref() + .is_some_and(|expected| expected != &owner) + { return Err(Error::NameHasNoOwner(destination)); } message.header.expected_responder = Some(owner.clone()); @@ -794,12 +821,20 @@ impl Connection { } else { None } - } else { None }; + } else { + None + }; let (tx, rx) = oneshot::channel(); - self.inner.pending.lock().await.insert(serial, PendingReply { - sender: tx, expected, bus_control, confidential: message.header.confidential, - }); + self.inner.pending.lock().await.insert( + serial, + PendingReply { + sender: tx, + expected, + bus_control, + confidential: message.header.confidential, + }, + ); if let Err(e) = self.enqueue(message).await { self.inner.pending.lock().await.remove(&serial); @@ -1267,8 +1302,13 @@ async fn dispatch_loop(inner: Arc) { // An unmatched reply is a timed-out call arriving late. Normal, // and nothing to do but drop it. let mut pending = inner.pending.lock().await; - if pending.get(&serial).is_some_and(|waiter| waiter.accepts(&message, brokered)) { - if let Some(waiter) = pending.remove(&serial) { let _ = waiter.sender.send(message); } + if pending + .get(&serial) + .is_some_and(|waiter| waiter.accepts(&message, brokered)) + { + if let Some(waiter) = pending.remove(&serial) { + let _ = waiter.sender.send(message); + } } } MessageKind::Signal => { diff --git a/crates/tinybus/src/connection_reply_tests.rs b/crates/tinybus/src/connection_reply_tests.rs index 20e8b1dc..425d6356 100644 --- a/crates/tinybus/src/connection_reply_tests.rs +++ b/crates/tinybus/src/connection_reply_tests.rs @@ -13,54 +13,99 @@ async fn another_broker_peer_cannot_complete_a_waiter_with_a_forged_reply() { BusName::new(crate::BUS_NAME).unwrap(), ObjectPath::new(crate::BUS_PATH).unwrap(), InterfaceName::new(crate::BUS_INTERFACE).unwrap(), - MemberName::new("Hello").unwrap(), serde_json::json!([]), + MemberName::new("Hello").unwrap(), + serde_json::json!([]), ); hello.header.serial = 1; responder.send(hello).await.unwrap(); - let responder_name = BusName::new(responder.recv().await.unwrap().unwrap().body.as_str().unwrap()).unwrap(); - let requester = Connection::connect(bus.connect().await.unwrap()).await.unwrap(); - let forger = Connection::connect(bus.connect().await.unwrap()).await.unwrap(); - let mut barriers = requester.add_match(MatchRule::new().signals()).await.unwrap(); + let responder_name = BusName::new( + responder + .recv() + .await + .unwrap() + .unwrap() + .body + .as_str() + .unwrap(), + ) + .unwrap(); + let requester = Connection::connect(bus.connect().await.unwrap()) + .await + .unwrap(); + let forger = Connection::connect(bus.connect().await.unwrap()) + .await + .unwrap(); + let mut barriers = requester + .add_match(MatchRule::new().signals()) + .await + .unwrap(); let request = Message::method_call( - responder_name.clone(), ObjectPath::new("/fixture").unwrap(), + responder_name.clone(), + ObjectPath::new("/fixture").unwrap(), InterfaceName::new("ai.tinyhumans.Fixture").unwrap(), - MemberName::new("Await").unwrap(), serde_json::json!([]), + MemberName::new("Await").unwrap(), + serde_json::json!([]), ); - let pending = tokio::spawn({ let requester = requester.clone(); async move { requester.call_raw(request, DEFAULT_TIMEOUT).await } }); + let pending = tokio::spawn({ + let requester = requester.clone(); + async move { requester.call_raw(request, DEFAULT_TIMEOUT).await } + }); let delivered = responder.recv().await.unwrap().unwrap(); assert_eq!(delivered.header.sender, requester.unique_name()); let mut forged = Message::method_return(&delivered.header, serde_json::json!("forged")); // Even a claimed B sender is overwritten to the actual peer C. forged.header.sender = Some(responder_name.clone()); forger.send(forged).await.unwrap(); - forger.emit(ObjectPath::new("/fixture").unwrap(), InterfaceName::new("ai.tinyhumans.Fixture").unwrap(), MemberName::new("Barrier").unwrap(), ()).await.unwrap(); - let barrier = tokio::time::timeout(DEFAULT_TIMEOUT, barriers.recv()).await.unwrap().unwrap(); + forger + .emit( + ObjectPath::new("/fixture").unwrap(), + InterfaceName::new("ai.tinyhumans.Fixture").unwrap(), + MemberName::new("Barrier").unwrap(), + (), + ) + .await + .unwrap(); + let barrier = tokio::time::timeout(DEFAULT_TIMEOUT, barriers.recv()) + .await + .unwrap() + .unwrap(); assert_eq!(barrier.header.sender, forger.unique_name()); // That connection's ordered signal arrived after its forged response, so // requester dispatch has processed the forgery before B is allowed to reply. let mut genuine = Message::method_return(&delivered.header, serde_json::json!("genuine")); genuine.header.serial = 2; responder.send(genuine).await.unwrap(); - assert_eq!(pending.await.unwrap().unwrap(), serde_json::json!("genuine")); + assert_eq!( + pending.await.unwrap().unwrap(), + serde_json::json!("genuine") + ); } - fn fixture_call(destination: BusName) -> Message { - Message::method_call(destination, ObjectPath::new("/fixture").unwrap(), + Message::method_call( + destination, + ObjectPath::new("/fixture").unwrap(), InterfaceName::new("ai.tinyhumans.Fixture").unwrap(), - MemberName::new("Await").unwrap(), serde_json::json!([])) + MemberName::new("Await").unwrap(), + serde_json::json!([]), + ) } async fn raw_control(peer: &dyn Transport, serial: u64, member: &str, body: Value) -> Message { - let mut call = Message::method_call(BusName::new(crate::BUS_NAME).unwrap(), + let mut call = Message::method_call( + BusName::new(crate::BUS_NAME).unwrap(), ObjectPath::new(crate::BUS_PATH).unwrap(), InterfaceName::new(crate::BUS_INTERFACE).unwrap(), - MemberName::new(member).unwrap(), body); + MemberName::new(member).unwrap(), + body, + ); call.header.serial = serial; peer.send(call).await.unwrap(); loop { let reply = peer.recv().await.unwrap().unwrap(); - if reply.header.reply_serial == Some(serial) { return reply; } + if reply.header.reply_serial == Some(serial) { + return reply; + } } } @@ -73,25 +118,48 @@ async fn reject_forged_replies(named: bool, confidential: bool, genuine_error: b let responder_name = BusName::new(hello.body.as_str().unwrap()).unwrap(); let alias = BusName::new("ai.tinyhumans.Fixture").unwrap(); if named { - let reply = raw_control(responder.as_ref(), 2, "RequestName", serde_json::json!([alias])).await; + let reply = raw_control( + responder.as_ref(), + 2, + "RequestName", + serde_json::json!([alias]), + ) + .await; assert_eq!(reply.header.kind, MessageKind::MethodReturn); } #[cfg(feature = "modules")] if confidential { - broker.attest_module(&responder_name, crate::attest::Attestation { - name: alias.clone(), sha256: "a".repeat(64), - }); + broker.attest_module( + &responder_name, + crate::attest::Attestation { + name: alias.clone(), + sha256: "a".repeat(64), + }, + ); } - let requester = Connection::connect(bus.connect().await.unwrap()).await.unwrap(); - let forger = Connection::connect(bus.connect().await.unwrap()).await.unwrap(); - let mut barriers = requester.add_match(MatchRule::new().signals()).await.unwrap(); + let requester = Connection::connect(bus.connect().await.unwrap()) + .await + .unwrap(); + let forger = Connection::connect(bus.connect().await.unwrap()) + .await + .unwrap(); + let mut barriers = requester + .add_match(MatchRule::new().signals()) + .await + .unwrap(); let mut request = fixture_call(if named { alias } else { responder_name.clone() }); request.header.confidential = confidential; - let pending = tokio::spawn({ let requester = requester.clone(); async move { - requester.call_raw(request, DEFAULT_TIMEOUT).await - }}); + let pending = tokio::spawn({ + let requester = requester.clone(); + async move { requester.call_raw(request, DEFAULT_TIMEOUT).await } + }); let delivered = responder.recv().await.unwrap().unwrap(); - if named { assert_eq!(delivered.header.expected_responder, Some(responder_name.clone())); } + if named { + assert_eq!( + delivered.header.expected_responder, + Some(responder_name.clone()) + ); + } for error in [false, true] { for private in [false, true] { let mut forged = if error { @@ -100,20 +168,33 @@ async fn reject_forged_replies(named: bool, confidential: bool, genuine_error: b Message::method_return(&delivered.header, serde_json::json!("forged")) }; forged.header.confidential = private; - forged.header.sensitive = private; + forged.header.sensitive = false; forged.header.sender = Some(BusName::new(crate::BUS_NAME).unwrap()); forger.send(forged).await.unwrap(); } } - forger.emit(ObjectPath::new("/fixture").unwrap(), - InterfaceName::new("ai.tinyhumans.Fixture").unwrap(), - MemberName::new("Barrier").unwrap(), ()).await.unwrap(); - let barrier = tokio::time::timeout(DEFAULT_TIMEOUT, barriers.recv()).await.unwrap().unwrap(); + forger + .emit( + ObjectPath::new("/fixture").unwrap(), + InterfaceName::new("ai.tinyhumans.Fixture").unwrap(), + MemberName::new("Barrier").unwrap(), + (), + ) + .await + .unwrap(); + let barrier = tokio::time::timeout(DEFAULT_TIMEOUT, barriers.recv()) + .await + .unwrap() + .unwrap(); assert_eq!(barrier.header.sender, forger.unique_name()); let mut genuine = if genuine_error { - Message::error_reply(&delivered.header, &Error::UnknownMethod { - interface: InterfaceName::new("ai.tinyhumans.Fixture").unwrap(), member: MemberName::new("Await").unwrap(), - }) + Message::error_reply( + &delivered.header, + &Error::UnknownMethod { + interface: InterfaceName::new("ai.tinyhumans.Fixture").unwrap(), + member: MemberName::new("Await").unwrap(), + }, + ) } else { Message::method_return(&delivered.header, serde_json::json!("genuine")) }; @@ -124,7 +205,9 @@ async fn reject_forged_replies(named: bool, confidential: bool, genuine_error: b let error = result.unwrap_err(); assert!(matches!(error, Error::MethodFailed { ref name, .. } if name == Error::UnknownMethod { interface: InterfaceName::new("ai.tinyhumans.Fixture").unwrap(), member: MemberName::new("Await").unwrap() }.wire_name())); - } else { assert_eq!(result.unwrap(), serde_json::json!("genuine")); } + } else { + assert_eq!(result.unwrap(), serde_json::json!("genuine")); + } } #[tokio::test] @@ -153,17 +236,31 @@ async fn confidential_unique_reply_preserves_genuine_sanitized_error() { async fn captured_alias_owner_is_checked_atomically_before_dispatch() { let bus = MemoryBus::new(); Broker::new().spawn(bus.clone()); - let original = Connection::connect(bus.connect().await.unwrap()).await.unwrap(); + let original = Connection::connect(bus.connect().await.unwrap()) + .await + .unwrap(); let replacement = bus.connect().await.unwrap(); let requester = bus.connect().await.unwrap(); raw_control(replacement.as_ref(), 1, "Hello", serde_json::json!([])).await; raw_control(requester.as_ref(), 1, "Hello", serde_json::json!([])).await; let alias = BusName::new("ai.tinyhumans.Fixture").unwrap(); original.request_name(alias.clone()).await.unwrap(); - let captured_reply = raw_control(requester.as_ref(), 2, "ResolveCallTarget", serde_json::json!([alias])).await; + let captured_reply = raw_control( + requester.as_ref(), + 2, + "ResolveCallTarget", + serde_json::json!([alias]), + ) + .await; let captured: BusName = serde_json::from_value(captured_reply.body).unwrap(); original.release_name(alias.clone()).await.unwrap(); - raw_control(replacement.as_ref(), 2, "RequestName", serde_json::json!([alias])).await; + raw_control( + replacement.as_ref(), + 2, + "RequestName", + serde_json::json!([alias]), + ) + .await; let mut call = fixture_call(alias.clone()); call.header.expected_responder = Some(captured.clone()); call.header.serial = 3; @@ -171,8 +268,14 @@ async fn captured_alias_owner_is_checked_atomically_before_dispatch() { let refusal = requester.recv().await.unwrap().unwrap(); assert_eq!(refusal.header.reply_serial, Some(3)); assert_eq!(refusal.header.kind, MessageKind::Error); - assert_eq!(refusal.header.sender.as_ref().unwrap().as_str(), crate::BUS_NAME); - assert_eq!(refusal.header.error_name.as_deref(), Some(Error::NAME_HAS_NO_OWNER)); + assert_eq!( + refusal.header.sender.as_ref().unwrap().as_str(), + crate::BUS_NAME + ); + assert_eq!( + refusal.header.error_name.as_deref(), + Some(Error::NameHasNoOwner(alias.clone()).wire_name()) + ); // An ordered broker control reply to the replacement is a barrier after // the refused call: an incorrectly routed call would arrive before it. let mut ping = fixture_call(BusName::new(crate::BUS_NAME).unwrap()); @@ -184,7 +287,9 @@ async fn captured_alias_owner_is_checked_atomically_before_dispatch() { let first = replacement.recv().await.unwrap().unwrap(); assert_eq!(first.header.kind, MessageKind::MethodReturn); assert_eq!(first.header.reply_serial, Some(3)); - let sdk = Connection::connect(bus.connect().await.unwrap()).await.unwrap(); + let sdk = Connection::connect(bus.connect().await.unwrap()) + .await + .unwrap(); let mut stale = fixture_call(alias.clone()); stale.header.expected_responder = Some(captured); assert!(matches!(sdk.call_raw(stale, DEFAULT_TIMEOUT).await, @@ -196,12 +301,147 @@ async fn raw_direct_reply_compatibility_does_not_claim_broker_identity() { let (a, b) = crate::transport::memory::MemoryTransport::pair(); let requester = Connection::attach(Arc::new(a)); let responder = Arc::new(b); - let pending = tokio::spawn({ let requester = requester.clone(); async move { - requester.call_raw(fixture_call(BusName::new("ai.tinyhumans.Fixture").unwrap()), DEFAULT_TIMEOUT).await - }}); + let pending = tokio::spawn({ + let requester = requester.clone(); + async move { + requester + .call_raw( + fixture_call(BusName::new("ai.tinyhumans.Fixture").unwrap()), + DEFAULT_TIMEOUT, + ) + .await + } + }); let delivered = responder.recv().await.unwrap().unwrap(); let mut response = Message::method_return(&delivered.header, serde_json::json!("raw")); response.header.serial = 1; responder.send(response).await.unwrap(); assert_eq!(pending.await.unwrap().unwrap(), serde_json::json!("raw")); } + +#[test] +fn reply_authority_rejects_other_peers_and_requires_private_success() { + let expected = BusName::new(":1.2").unwrap(); + let caller = BusName::new(":1.1").unwrap(); + let (sender, _) = oneshot::channel(); + let waiter = PendingReply { + sender, + expected: Some(expected.clone()), + bus_control: false, + confidential: true, + }; + let mut call = fixture_call(expected.clone()); + call.header.serial = 8; + call.header.sender = Some(caller); + for kind in [MessageKind::MethodReturn, MessageKind::Error] { + for private in [false, true] { + for sensitive in [false, true] { + let mut reply = Message::method_return(&call.header, Value::Null); + reply.header.kind = kind; + reply.header.sender = Some(BusName::new(":1.3").unwrap()); + reply.header.confidential = private; + reply.header.sensitive = sensitive; + assert!(!waiter.accepts(&reply, true)); + reply.header.sender = Some(expected.clone()); + assert_eq!( + waiter.accepts(&reply, true), + kind == MessageKind::Error || private + ); + reply.header.sender = Some(BusName::new(crate::BUS_NAME).unwrap()); + assert_eq!(waiter.accepts(&reply, true), kind == MessageKind::Error); + } + } + } + let (sender, _) = oneshot::channel(); + let control = PendingReply { + sender, + expected: None, + bus_control: true, + confidential: false, + }; + let mut reply = Message::method_return(&call.header, Value::Null); + reply.header.sender = Some(expected); + assert!(!control.accepts(&reply, true)); + reply.header.sender = None; + assert!(control.accepts(&reply, true)); +} + +#[tokio::test] +async fn legacy_alias_calls_remain_routable_without_a_conditional_header() { + let bus = MemoryBus::new(); + Broker::new().spawn(bus.clone()); + let responder = bus.connect().await.unwrap(); + raw_control(responder.as_ref(), 1, "Hello", serde_json::json!([])).await; + let alias = BusName::new("ai.tinyhumans.Fixture").unwrap(); + raw_control( + responder.as_ref(), + 2, + "RequestName", + serde_json::json!([alias]), + ) + .await; + let requester = bus.connect().await.unwrap(); + raw_control(requester.as_ref(), 1, "Hello", serde_json::json!([])).await; + let mut call = fixture_call(alias); + call.header.serial = 2; + assert!(call.header.expected_responder.is_none()); + requester.send(call).await.unwrap(); + let delivered = responder.recv().await.unwrap().unwrap(); + assert_eq!(delivered.header.kind, MessageKind::MethodCall); + let mut reply = Message::method_return(&delivered.header, serde_json::json!("legacy")); + reply.header.serial = 3; + responder.send(reply).await.unwrap(); + assert_eq!( + requester.recv().await.unwrap().unwrap().body, + serde_json::json!("legacy") + ); +} + +#[tokio::test] +async fn unsupported_owner_guard_negotiation_fails_before_alias_delivery() { + let (a, b) = crate::transport::memory::MemoryTransport::pair(); + let server = tokio::spawn(async move { + let hello = b.recv().await.unwrap().unwrap(); + let mut hello_reply = Message::method_return(&hello.header, serde_json::json!(":1.1")); + hello_reply.header.serial = 1; + b.send(hello_reply).await.unwrap(); + let lookup = b.recv().await.unwrap().unwrap(); + assert_eq!( + lookup.header.member.as_ref().unwrap().as_str(), + "ResolveCallTarget" + ); + let mut refusal = Message::error_reply( + &lookup.header, + &Error::UnknownMethod { + interface: InterfaceName::new(crate::BUS_INTERFACE).unwrap(), + member: MemberName::new("ResolveCallTarget").unwrap(), + }, + ); + refusal.header.serial = 2; + b.send(refusal).await.unwrap(); + }); + let requester = Connection::connect(Box::new(a)).await.unwrap(); + let error = requester + .call_raw( + fixture_call(BusName::new("ai.tinyhumans.Fixture").unwrap()), + DEFAULT_TIMEOUT, + ) + .await + .unwrap_err(); + assert!(matches!(error, Error::MethodFailed { ref name, .. } if name == Error::UNKNOWN_METHOD)); + server.await.unwrap(); +} + +#[test] +fn conditional_owner_header_requires_a_unique_method_call_target() { + let mut call = fixture_call(BusName::new("ai.tinyhumans.Fixture").unwrap()); + call.header.serial = 1; + call.header.expected_responder = Some(BusName::new("ai.tinyhumans.Other").unwrap()); + assert!(call.validate().is_err()); + call.header.expected_responder = Some(BusName::new(":1.2").unwrap()); + assert!(call.validate().is_ok()); + let mut reply = Message::method_return(&call.header, Value::Null); + reply.header.serial = 2; + reply.header.expected_responder = call.header.expected_responder; + assert!(reply.validate().is_err()); +} diff --git a/crates/tinybus/src/message/mod.rs b/crates/tinybus/src/message/mod.rs index 46891149..d8fad8ed 100644 --- a/crates/tinybus/src/message/mod.rs +++ b/crates/tinybus/src/message/mod.rs @@ -338,7 +338,9 @@ impl Message { if let Some(expected) = &self.header.expected_responder && (self.header.kind != MessageKind::MethodCall || !expected.is_unique()) { - return Err(Error::protocol("expected responder must be unique and only accompany a call")); + return Err(Error::protocol( + "expected responder must be unique and only accompany a call", + )); } // Checked before the per-kind rules, and checked on ingress rather than // at delivery: a confidential signal has no destination, so there is no diff --git a/crates/tinybus/src/router.rs b/crates/tinybus/src/router.rs index ec148874..6e2fddc5 100644 --- a/crates/tinybus/src/router.rs +++ b/crates/tinybus/src/router.rs @@ -494,7 +494,11 @@ impl Router { if expected.is_some_and(|expected| self.owner_of(destination).as_ref() != Some(expected)) { return Err(Error::NameHasNoOwner(destination.clone())); } - if confidential { self.resolve_attested(destination) } else { self.resolve(destination) } + if confidential { + self.resolve_attested(destination) + } else { + self.resolve(destination) + } } /// The outbox of whoever owns `destination`. diff --git a/target-reply-tests.log b/target-reply-tests.log index 4e0fb047..3d9dabdc 100644 --- a/target-reply-tests.log +++ b/target-reply-tests.log @@ -1,15 +1,3 @@ [ci-cancel-aware] exec: cargo test -p tinybus --all-features reply_tests Compiling tinybus v0.1.3 (/home/enamakel/work/openhuman/worktrees/cloud-storage-7419/vendor/tinybus/crates/tinybus) -error[E0599]: no variant, associated function, or constant named `NAME_HAS_NO_OWNER` found for enum `error::Error` in the current scope - --> crates/tinybus/src/connection_reply_tests.rs:175:66 - | -175 | assert_eq!(refusal.header.error_name.as_deref(), Some(Error::NAME_HAS_NO_OWNER)); - | ^^^^^^^^^^^^^^^^^ variant, associated function, or constant not found in `error::Error` - | - ::: crates/tinybus/src/error.rs:22:1 - | - 22 | pub enum Error { - | -------------- variant, associated function, or constant `NAME_HAS_NO_OWNER` not found for this enum - -For more information about this error, try `rustc --explain E0599`. -error: could not compile `tinybus` (lib test) due to 1 previous error + Compiling tinybus-module v0.1.3 (/home/enamakel/work/openhuman/worktrees/cloud-storage-7419/vendor/tinybus/crates/tinybus-module) From 89c1f2fb1da86eee773c9927edf524187819957a Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 11 Oct 2026 19:23:51 +0300 Subject: [PATCH 07/14] chore: add reply test run output to log Record the successful cargo test run for the tinybus reply_tests suite, showing all 11 tests passing, so the captured output is available alongside the existing log. Auto-committed-on: dragonfly Co-authored-by: Medulla --- target-reply-tests.log | 24 ++++++++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/target-reply-tests.log b/target-reply-tests.log index 3d9dabdc..108a441b 100644 --- a/target-reply-tests.log +++ b/target-reply-tests.log @@ -1,3 +1,27 @@ [ci-cancel-aware] exec: cargo test -p tinybus --all-features reply_tests Compiling tinybus v0.1.3 (/home/enamakel/work/openhuman/worktrees/cloud-storage-7419/vendor/tinybus/crates/tinybus) Compiling tinybus-module v0.1.3 (/home/enamakel/work/openhuman/worktrees/cloud-storage-7419/vendor/tinybus/crates/tinybus-module) + Finished `test` profile [unoptimized + debuginfo] target(s) in 19.84s + Running unittests src/lib.rs (target/debug/deps/tinybus-b3a640fb890222d1) + +running 11 tests +test connection::reply_tests::conditional_owner_header_requires_a_unique_method_call_target ... ok +test connection::reply_tests::reply_authority_rejects_other_peers_and_requires_private_success ... ok +test connection::reply_tests::raw_direct_reply_compatibility_does_not_claim_broker_identity ... ok +test connection::reply_tests::legacy_alias_calls_remain_routable_without_a_conditional_header ... ok +test connection::reply_tests::unsupported_owner_guard_negotiation_fails_before_alias_delivery ... ok +test connection::reply_tests::another_broker_peer_cannot_complete_a_waiter_with_a_forged_reply ... ok +test connection::reply_tests::confidential_unique_reply_preserves_genuine_sanitized_error ... ok +test connection::reply_tests::confidential_named_reply_binds_the_attested_owner ... ok +test connection::reply_tests::unique_reply_preserves_genuine_error_after_rejecting_forged_errors ... ok +test connection::reply_tests::captured_alias_owner_is_checked_atomically_before_dispatch ... ok +test connection::reply_tests::named_reply_rejects_forged_success_and_errors_with_any_private_flags ... ok + +test result: ok. 11 passed; 0 failed; 0 ignored; 0 measured; 473 filtered out; finished in 0.00s + + Running unittests src/bin/tinybus.rs (target/debug/deps/tinybus-50b3c4f41422dede) + +running 0 tests + +test result: ok. 0 passed; 0 failed; 0 ignored; 0 measured; 6 filtered out; finished in 0.00s + From d1f3855bafd29a51bcde9eb4aeb3005664debcac Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 11 Oct 2026 19:24:54 +0300 Subject: [PATCH 08/14] test(tinybus): cover confidential alias takeover and legacy wire frames Refactor the alias takeover test into a shared helper so it can run both with and without the modules feature, and add a confidential variant that attests the original owner before the call. Also add tests for native wire names on genuine broker errors and for round-tripping frames that omit the optional expected_responder header, and document the conditional routing and ResolveCallTarget negotiation in the protocol spec. Auto-committed-on: dragonfly Co-authored-by: Medulla --- crates/tinybus/src/connection_reply_tests.rs | 68 +++++++++++++++++++- docs/protocol.md | 28 ++++++++ target-forged-reply-green.log | 12 ---- target-forged-reply-red.log | 68 -------------------- target-reply-tests.log | 27 -------- 5 files changed, 93 insertions(+), 110 deletions(-) delete mode 100644 target-forged-reply-green.log delete mode 100644 target-forged-reply-red.log delete mode 100644 target-reply-tests.log diff --git a/crates/tinybus/src/connection_reply_tests.rs b/crates/tinybus/src/connection_reply_tests.rs index 425d6356..3aad114d 100644 --- a/crates/tinybus/src/connection_reply_tests.rs +++ b/crates/tinybus/src/connection_reply_tests.rs @@ -232,10 +232,10 @@ async fn confidential_unique_reply_preserves_genuine_sanitized_error() { reject_forged_replies(false, true, true).await; } -#[tokio::test] -async fn captured_alias_owner_is_checked_atomically_before_dispatch() { +async fn alias_takeover_is_refused(confidential: bool) { let bus = MemoryBus::new(); - Broker::new().spawn(bus.clone()); + let broker = Broker::new(); + broker.clone().spawn(bus.clone()); let original = Connection::connect(bus.connect().await.unwrap()) .await .unwrap(); @@ -245,6 +245,16 @@ async fn captured_alias_owner_is_checked_atomically_before_dispatch() { raw_control(requester.as_ref(), 1, "Hello", serde_json::json!([])).await; let alias = BusName::new("ai.tinyhumans.Fixture").unwrap(); original.request_name(alias.clone()).await.unwrap(); + #[cfg(feature = "modules")] + if confidential { + broker.attest_module( + &original.unique_name().unwrap(), + crate::attest::Attestation { + name: alias.clone(), + sha256: "a".repeat(64), + }, + ); + } let captured_reply = raw_control( requester.as_ref(), 2, @@ -263,6 +273,7 @@ async fn captured_alias_owner_is_checked_atomically_before_dispatch() { .await; let mut call = fixture_call(alias.clone()); call.header.expected_responder = Some(captured.clone()); + call.header.confidential = confidential; call.header.serial = 3; requester.send(call).await.unwrap(); let refusal = requester.recv().await.unwrap().unwrap(); @@ -445,3 +456,54 @@ fn conditional_owner_header_requires_a_unique_method_call_target() { reply.header.expected_responder = call.header.expected_responder; assert!(reply.validate().is_err()); } + +#[tokio::test] +async fn captured_alias_owner_is_checked_atomically_before_dispatch() { + alias_takeover_is_refused(false).await; +} + +#[cfg(feature = "modules")] +#[tokio::test] +async fn confidential_alias_takeover_is_refused_before_dispatch() { + alias_takeover_is_refused(true).await; +} + +#[tokio::test] +async fn genuine_broker_routing_and_control_errors_retain_native_wire_names() { + let bus = MemoryBus::new(); + Broker::new().spawn(bus.clone()); + let requester = Connection::connect(bus.connect().await.unwrap()) + .await + .unwrap(); + let missing = BusName::new(":1.9999").unwrap(); + let error = requester + .call_raw(fixture_call(missing.clone()), DEFAULT_TIMEOUT) + .await + .unwrap_err(); + assert!(matches!(error, Error::MethodFailed { ref name, .. } + if name == Error::NameHasNoOwner(missing).wire_name())); + let mut private = fixture_call(BusName::new(crate::BUS_NAME).unwrap()); + private.header.confidential = true; + let error = requester + .call_raw(private, DEFAULT_TIMEOUT) + .await + .unwrap_err(); + assert!(matches!(error, Error::MethodFailed { ref name, .. } + if name == Error::not_attested(BusName::new(crate::BUS_NAME).unwrap(), "fixture").wire_name())); +} + +#[test] +fn absent_conditional_header_round_trips_legacy_wire_frames() { + let mut call = fixture_call(BusName::new("ai.tinyhumans.Fixture").unwrap()); + call.header.serial = 1; + let encoded = serde_json::to_value(&call).unwrap(); + assert!(encoded["header"].get("expected_responder").is_none()); + let decoded: Message = serde_json::from_value(encoded).unwrap(); + assert!(decoded.header.expected_responder.is_none()); + call.header.expected_responder = Some(BusName::new(":1.2").unwrap()); + let decoded: Message = serde_json::from_value(serde_json::to_value(call).unwrap()).unwrap(); + assert_eq!( + decoded.header.expected_responder.as_ref().unwrap().as_str(), + ":1.2" + ); +} diff --git a/docs/protocol.md b/docs/protocol.md index 1645df35..de2f5973 100644 --- a/docs/protocol.md +++ b/docs/protocol.md @@ -50,6 +50,7 @@ omitted field and a `null` field identically. | `serial` | u64, ≥ 1, per-sender monotonic | always | | `reply_serial` | u64 | `method_return`, `error` | | `sender` | bus name | stamped by the broker on everything it forwards | +| `expected_responder` | unique bus name | optional, conditional method-call owner | | `destination` | bus name | `method_call`, `method_return`, `error` | | `path` | object path | `method_call`, `signal` | | `interface` | interface name | `method_call`, `signal` | @@ -66,6 +67,32 @@ the error name, which travels in `error_name`. A peer **may** set `sender`; the broker overwrites it unconditionally. Nothing downstream may trust a `sender` that did not come from the broker. +Replies must match both the pending serial and its captured unique responder. +Only errors stamped by the broker itself may substitute for that responder. +Control calls accept the broker's reserved sender or its legacy unstamped +control replies; forwarded peer replies always have a unique sender. A +confidential call requires a confidential successful reply; sanitized errors +remain unflagged. + +For named calls, the SDK first invokes `ResolveCallTarget`, then sets +`expected_responder` to the returned unique owner. The broker compares this +guard with the current owner while selecting the recipient and checking +confidential admission under one routing lock. A changed owner produces +`NameHasNoOwner` without dispatch. Ownership can change after selection: +the already selected peer remains the only accepted responder. + +Old clients omit the optional field and continue routing on a new broker. +New clients refuse named calls when an old broker lacks `ResolveCallTarget`; +they retain the typed `UnknownMethod` error rather than silently sending an +unguarded call. Unique calls need no alias negotiation. + +These identity guarantees require the actual broker to overwrite ingress +senders and protect its transport. A raw `Connection::attach` pair without +the broker handshake retains unverified direct-call compatibility. Payload +headers on a raw link are not authenticated, and this API does not establish +remote peer identity or turn a custom transport into a trusted broker. + + ### `confidential` A peer **may** set `confidential`, and the broker does **not** overwrite it. The @@ -155,6 +182,7 @@ interface `ai.tinyhumans.tinybus.Bus`. | `GetManifest` | `[name]` | that peer's manifest, or `null` | | `ListPeers` | `[]` | unique names, owned names, and peer manifests | | `GetNameOwner` | `[name]` | the owner's unique name, or `null` | +| `ResolveCallTarget` | `[name]` | unique owner, or `NameHasNoOwner`; negotiates conditional routing | | `GetAttestation` | `[name]` | what the host verified about that owner, or `null` | | `AddMatch` | `[rule]` | `null` | | `RemoveMatch` | `[rule]` | `null` | diff --git a/target-forged-reply-green.log b/target-forged-reply-green.log deleted file mode 100644 index 05266073..00000000 --- a/target-forged-reply-green.log +++ /dev/null @@ -1,12 +0,0 @@ -[ci-cancel-aware] exec: cargo test -p tinybus another_broker_peer_cannot_complete - Blocking waiting for file lock on package cache - Compiling tinybus v0.1.3 (/home/enamakel/work/openhuman/worktrees/cloud-storage-7419/vendor/tinybus/crates/tinybus) - Compiling tinybus-module v0.1.3 (/home/enamakel/work/openhuman/worktrees/cloud-storage-7419/vendor/tinybus/crates/tinybus-module) - Finished `test` profile [unoptimized + debuginfo] target(s) in 16.40s - Running unittests src/lib.rs (target/debug/deps/tinybus-49c252397df0a704) - -running 1 test -test connection::reply_tests::another_broker_peer_cannot_complete_a_waiter_with_a_forged_reply ... ok - -test result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 282 filtered out; finished in 0.00s - diff --git a/target-forged-reply-red.log b/target-forged-reply-red.log deleted file mode 100644 index d4f692a3..00000000 --- a/target-forged-reply-red.log +++ /dev/null @@ -1,68 +0,0 @@ -[ci-cancel-aware] exec: cargo test -p tinybus another_broker_peer_cannot_complete - Compiling proc-macro2 v1.0.107 - Compiling quote v1.0.47 - Compiling unicode-ident v1.0.24 - Compiling libc v0.2.189 - Compiling version_check v0.9.5 - Compiling typenum v1.20.1 - Compiling generic-array v0.14.7 - Compiling serde_core v1.0.229 - Compiling cfg-if v1.0.4 - Compiling once_cell v1.21.4 - Compiling zmij v1.0.23 - Compiling thiserror v2.0.20 - Compiling serde_json v1.0.151 - Compiling block-buffer v0.10.4 - Compiling crypto-common v0.1.7 - Compiling pin-project-lite v0.2.17 - Compiling serde v1.0.229 - Compiling digest v0.10.7 - Compiling syn v3.0.3 - Compiling syn v2.0.119 - Compiling tracing-core v0.1.36 - Compiling socket2 v0.6.5 - Compiling thiserror-impl v2.0.20 - Compiling tracing-attributes v0.1.31 - Compiling tokio-macros v2.7.2 - Compiling serde_derive v1.0.229 - Compiling mio v1.2.2 - Compiling itoa v1.0.18 - Compiling cpufeatures v0.2.17 - Compiling rustix v1.1.4 - Compiling memchr v2.8.3 - Compiling tinybus v0.1.3 (/home/enamakel/work/openhuman/worktrees/cloud-storage-7419/vendor/tinybus/crates/tinybus) - Compiling getrandom v0.4.3 - Compiling bytes v1.12.1 - Compiling sha2 v0.10.9 - Compiling tokio v1.53.1 - Compiling tracing v0.1.44 - Compiling tinybus-macros v0.1.3 (/home/enamakel/work/openhuman/worktrees/cloud-storage-7419/vendor/tinybus/crates/tinybus-macros) - Compiling async-trait v0.1.92 - Compiling linux-raw-sys v0.12.1 - Compiling bitflags v2.13.1 - Compiling fastrand v2.5.0 - Compiling tempfile v3.27.0 - Compiling tinybus-module v0.1.3 (/home/enamakel/work/openhuman/worktrees/cloud-storage-7419/vendor/tinybus/crates/tinybus-module) - Finished `test` profile [unoptimized + debuginfo] target(s) in 28.09s - Running unittests src/lib.rs (target/debug/deps/tinybus-49c252397df0a704) - -running 1 test -test connection::reply_tests::another_broker_peer_cannot_complete_a_waiter_with_a_forged_reply ... FAILED - -failures: - ----- connection::reply_tests::another_broker_peer_cannot_complete_a_waiter_with_a_forged_reply stdout ---- - -thread 'connection::reply_tests::another_broker_peer_cannot_complete_a_waiter_with_a_forged_reply' (3898707) panicked at crates/tinybus/src/connection_reply_tests.rs:42:5: -assertion `left == right` failed - left: String("forged") - right: String("genuine") -note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace - - -failures: - connection::reply_tests::another_broker_peer_cannot_complete_a_waiter_with_a_forged_reply - -test result: FAILED. 0 passed; 1 failed; 0 ignored; 0 measured; 282 filtered out; finished in 0.00s - -error: test failed, to rerun pass `-p tinybus --lib` diff --git a/target-reply-tests.log b/target-reply-tests.log deleted file mode 100644 index 108a441b..00000000 --- a/target-reply-tests.log +++ /dev/null @@ -1,27 +0,0 @@ -[ci-cancel-aware] exec: cargo test -p tinybus --all-features reply_tests - Compiling tinybus v0.1.3 (/home/enamakel/work/openhuman/worktrees/cloud-storage-7419/vendor/tinybus/crates/tinybus) - Compiling tinybus-module v0.1.3 (/home/enamakel/work/openhuman/worktrees/cloud-storage-7419/vendor/tinybus/crates/tinybus-module) - Finished `test` profile [unoptimized + debuginfo] target(s) in 19.84s - Running unittests src/lib.rs (target/debug/deps/tinybus-b3a640fb890222d1) - -running 11 tests -test connection::reply_tests::conditional_owner_header_requires_a_unique_method_call_target ... ok -test connection::reply_tests::reply_authority_rejects_other_peers_and_requires_private_success ... ok -test connection::reply_tests::raw_direct_reply_compatibility_does_not_claim_broker_identity ... ok -test connection::reply_tests::legacy_alias_calls_remain_routable_without_a_conditional_header ... ok -test connection::reply_tests::unsupported_owner_guard_negotiation_fails_before_alias_delivery ... ok -test connection::reply_tests::another_broker_peer_cannot_complete_a_waiter_with_a_forged_reply ... ok -test connection::reply_tests::confidential_unique_reply_preserves_genuine_sanitized_error ... ok -test connection::reply_tests::confidential_named_reply_binds_the_attested_owner ... ok -test connection::reply_tests::unique_reply_preserves_genuine_error_after_rejecting_forged_errors ... ok -test connection::reply_tests::captured_alias_owner_is_checked_atomically_before_dispatch ... ok -test connection::reply_tests::named_reply_rejects_forged_success_and_errors_with_any_private_flags ... ok - -test result: ok. 11 passed; 0 failed; 0 ignored; 0 measured; 473 filtered out; finished in 0.00s - - Running unittests src/bin/tinybus.rs (target/debug/deps/tinybus-50b3c4f41422dede) - -running 0 tests - -test result: ok. 0 passed; 0 failed; 0 ignored; 0 measured; 6 filtered out; finished in 0.00s - From 9ebe625e254bfd3a9126442c6af40f4a7ba62b3d Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 11 Oct 2026 19:25:31 +0300 Subject: [PATCH 09/14] fix(connection): accept unstamped replies on unverified direct transports Pending replies now record whether the connection has no unique name, and reply authority accepts a sender-less reply in that case as long as the delivery was not brokered. This restores raw direct transport compatibility for peers that cannot be authenticated, while still rejecting proven broker deliveries that lack a sender. Auto-committed-on: dragonfly Co-authored-by: Medulla --- crates/tinybus/src/connection.rs | 6 +++++- crates/tinybus/src/connection_reply_tests.rs | 7 +++---- 2 files changed, 8 insertions(+), 5 deletions(-) diff --git a/crates/tinybus/src/connection.rs b/crates/tinybus/src/connection.rs index 496c8dbd..6b1eb5a4 100644 --- a/crates/tinybus/src/connection.rs +++ b/crates/tinybus/src/connection.rs @@ -147,6 +147,7 @@ struct PendingReply { expected: Option, bus_control: bool, confidential: bool, + unverified_direct: bool, } impl PendingReply { @@ -158,7 +159,9 @@ impl PendingReply { // real broker can never send an unstamped frame. sender.is_none() || broker_sender } else if let Some(expected) = &self.expected { - sender == Some(expected) || (reply.header.kind == MessageKind::Error && broker_sender) + sender == Some(expected) + || (reply.header.kind == MessageKind::Error && broker_sender) + || (self.unverified_direct && !brokered && sender.is_none()) } else { // Raw direct transport compatibility has no authenticated peer // identity. A proven broker delivery must not use this escape. @@ -833,6 +836,7 @@ impl Connection { expected, bus_control, confidential: message.header.confidential, + unverified_direct: self.unique_name().is_none(), }, ); diff --git a/crates/tinybus/src/connection_reply_tests.rs b/crates/tinybus/src/connection_reply_tests.rs index 3aad114d..3f664650 100644 --- a/crates/tinybus/src/connection_reply_tests.rs +++ b/crates/tinybus/src/connection_reply_tests.rs @@ -316,10 +316,7 @@ async fn raw_direct_reply_compatibility_does_not_claim_broker_identity() { let requester = requester.clone(); async move { requester - .call_raw( - fixture_call(BusName::new("ai.tinyhumans.Fixture").unwrap()), - DEFAULT_TIMEOUT, - ) + .call_raw(fixture_call(BusName::new(":1.2").unwrap()), DEFAULT_TIMEOUT) .await } }); @@ -340,6 +337,7 @@ fn reply_authority_rejects_other_peers_and_requires_private_success() { expected: Some(expected.clone()), bus_control: false, confidential: true, + unverified_direct: false, }; let mut call = fixture_call(expected.clone()); call.header.serial = 8; @@ -369,6 +367,7 @@ fn reply_authority_rejects_other_peers_and_requires_private_success() { expected: None, bus_control: true, confidential: false, + unverified_direct: false, }; let mut reply = Message::method_return(&call.header, Value::Null); reply.header.sender = Some(expected); From b974b797d0ebacc8775e65f37ba073c03e00d7c1 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 11 Oct 2026 19:26:23 +0300 Subject: [PATCH 10/14] feat(broker): resolve call targets before taking the router lock ResolveCallTarget is now handled ahead of the main dispatch path so the router lock is not held while consulting the module lifecycle owner. The method also reports module-state errors for names with no live owner, matching the specificity of ordinary routing failures. Auto-committed-on: dragonfly Co-authored-by: Medulla --- crates/tinybus/src/broker.rs | 37 +++++++++++++++++++++++++++--------- 1 file changed, 28 insertions(+), 9 deletions(-) diff --git a/crates/tinybus/src/broker.rs b/crates/tinybus/src/broker.rs index dd3fe0e9..9eba9da6 100644 --- a/crates/tinybus/src/broker.rs +++ b/crates/tinybus/src/broker.rs @@ -335,6 +335,34 @@ impl Broker { return (result, Vec::new(), module_states); } + // Negotiate conditional routing before building the call body. Keep + // module-state errors as specific as ordinary routing, and do not hold + // the router lock while consulting the module lifecycle owner. + if member.as_str() == "ResolveCallTarget" { + let result = (|| -> Result { + let (name,): (BusName,) = parse_args(member, body)?; + let owner = self.router.lock().expect("router lock").owner_of(&name); + match owner { + Some(owner) => Ok(serde_json::to_value(owner)?), + None => { + #[cfg(feature = "modules")] + if let Some(error) = self + .modules + .lock() + .expect("module control lock") + .as_ref() + .and_then(Weak::upgrade) + .and_then(|control| control.unavailable_for(&name)) + { + return Err(error); + } + Err(Error::NameHasNoOwner(name)) + } + } + })(); + return (result, Vec::new(), Vec::new()); + } + let mut changes = Vec::new(); let mut router = self.router.lock().expect("router lock"); @@ -380,15 +408,6 @@ impl Broker { let (name,): (BusName,) = parse_args(member, body)?; Ok(serde_json::to_value(router.attestation_of(&name))?) } - // This method also negotiates conditional owner routing. An - // older broker fails loudly rather than ignoring the guard. - "ResolveCallTarget" => { - let (name,): (BusName,) = parse_args(member, body)?; - let owner = router - .owner_of(&name) - .ok_or_else(|| Error::NameHasNoOwner(name))?; - Ok(serde_json::to_value(owner)?) - } "GetNameOwner" => { let (name,): (BusName,) = parse_args(member, body)?; Ok(serde_json::to_value(router.owner_of(&name))?) From a0fe75c561ac3d3ad768dd2c0a4c6d7b5d7394e9 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 11 Oct 2026 19:29:55 +0300 Subject: [PATCH 11/14] fix(message): reject non-unique expected responder on non-call messages The expected-responder validation previously short-circuited when the field was absent, so the uniqueness and message-kind checks were skipped in some cases. The condition now evaluates the responder through is_some_and, ensuring the protocol error is raised whenever a present responder is non-unique or accompanies a non-call message. Auto-committed-on: dragonfly Co-authored-by: Medulla --- crates/tinybus/src/message/mod.rs | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/crates/tinybus/src/message/mod.rs b/crates/tinybus/src/message/mod.rs index d8fad8ed..1e3ec670 100644 --- a/crates/tinybus/src/message/mod.rs +++ b/crates/tinybus/src/message/mod.rs @@ -335,8 +335,13 @@ impl Message { "only a method call can request a streamed reply", )); } - if let Some(expected) = &self.header.expected_responder - && (self.header.kind != MessageKind::MethodCall || !expected.is_unique()) + if self + .header + .expected_responder + .as_ref() + .is_some_and(|expected| { + self.header.kind != MessageKind::MethodCall || !expected.is_unique() + }) { return Err(Error::protocol( "expected responder must be unique and only accompany a call", From 7298fd41d14aa66a373cab9fd1f4de2647145c15 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 11 Oct 2026 19:31:18 +0300 Subject: [PATCH 12/14] fix(connection): assign call serial after owner negotiation The outbound serial was allocated before validation and owner negotiation, so a guarded call could consume a serial ahead of the alias negotiation it depends on. The serial is now assigned only once the message is ready to go on the wire, and tests cover the ordering and the rejection of a non-unique responder. Auto-committed-on: dragonfly Co-authored-by: Medulla --- crates/tinybus/src/connection.rs | 7 ++- crates/tinybus/src/connection_reply_tests.rs | 52 ++++++++++++++++++++ 2 files changed, 57 insertions(+), 2 deletions(-) diff --git a/crates/tinybus/src/connection.rs b/crates/tinybus/src/connection.rs index 6b1eb5a4..d70fd873 100644 --- a/crates/tinybus/src/connection.rs +++ b/crates/tinybus/src/connection.rs @@ -765,8 +765,9 @@ impl Connection { } async fn call_raw_message(&self, mut message: Message, timeout: Duration) -> Result { - let serial = self.inner.serial.fetch_add(1, Ordering::Relaxed); - message.header.serial = serial; + // Validate without consuming the outbound serial: alias negotiation + // must precede the call on the wire and in the serial sequence. + message.header.serial = 1; message.validate()?; if message.header.confidential && message.header.stream_reply { return Err(Error::protocol( @@ -828,6 +829,8 @@ impl Connection { None }; + let serial = self.inner.serial.fetch_add(1, Ordering::Relaxed); + message.header.serial = serial; let (tx, rx) = oneshot::channel(); self.inner.pending.lock().await.insert( serial, diff --git a/crates/tinybus/src/connection_reply_tests.rs b/crates/tinybus/src/connection_reply_tests.rs index 3f664650..0359bd8d 100644 --- a/crates/tinybus/src/connection_reply_tests.rs +++ b/crates/tinybus/src/connection_reply_tests.rs @@ -506,3 +506,55 @@ fn absent_conditional_header_round_trips_legacy_wire_frames() { ":1.2" ); } + +async fn negotiated_owner_reply(owner: &str) -> Result { + let owner = owner.to_owned(); + let (a, b) = crate::transport::memory::MemoryTransport::pair(); + let server = tokio::spawn(async move { + let hello = b.recv().await.unwrap().unwrap(); + let mut reply = Message::method_return(&hello.header, serde_json::json!(":1.1")); + reply.header.serial = 1; + b.send(reply).await.unwrap(); + let lookup = b.recv().await.unwrap().unwrap(); + let mut reply = Message::method_return(&lookup.header, serde_json::json!(owner)); + reply.header.serial = 2; + b.send(reply).await.unwrap(); + if owner.starts_with(':') { + let call = b.recv().await.unwrap().unwrap(); + assert!(call.header.serial > lookup.header.serial); + assert_eq!( + call.header.expected_responder.as_ref().unwrap().as_str(), + owner + ); + let mut reply = Message::method_return(&call.header, serde_json::json!("bound")); + reply.header.sender = Some(BusName::new(owner).unwrap()); + reply.header.serial = 3; + b.send(reply).await.unwrap(); + } + }); + let requester = Connection::connect(Box::new(a)).await.unwrap(); + let result = requester + .call_raw( + fixture_call(BusName::new("ai.tinyhumans.Fixture").unwrap()), + DEFAULT_TIMEOUT, + ) + .await; + server.await.unwrap(); + result +} + +#[tokio::test] +async fn guarded_call_serial_follows_its_owner_negotiation() { + assert_eq!( + negotiated_owner_reply(":1.2").await.unwrap(), + serde_json::json!("bound") + ); +} + +#[tokio::test] +async fn owner_negotiation_rejects_non_unique_responder_before_delivery() { + assert!(matches!( + negotiated_owner_reply("ai.tinyhumans.Fixture").await, + Err(Error::Protocol(_)) + )); +} From a7bb72829301d4c059f1f131e40fcad1264e131d Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 11 Oct 2026 19:40:22 +0300 Subject: [PATCH 13/14] test(tinybus): cover stream ownership after name handover Add a regression test asserting that stream chunks and close stay with the peer that answered the open, even after the name is released and re-requested by another connection. Auto-committed-on: dragonfly Co-authored-by: Medulla --- crates/tinybus/src/connection_reply_tests.rs | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/crates/tinybus/src/connection_reply_tests.rs b/crates/tinybus/src/connection_reply_tests.rs index 0359bd8d..40b2d5df 100644 --- a/crates/tinybus/src/connection_reply_tests.rs +++ b/crates/tinybus/src/connection_reply_tests.rs @@ -558,3 +558,22 @@ async fn owner_negotiation_rejects_non_unique_responder_before_delivery() { Err(Error::Protocol(_)) )); } + +#[tokio::test] +async fn stream_chunks_and_close_stay_with_the_peer_that_answered_open() { + let bus = MemoryBus::new(); + Broker::new().spawn(bus.clone()); + let original = Connection::connect(bus.connect().await.unwrap()).await.unwrap(); + let replacement = Connection::connect(bus.connect().await.unwrap()).await.unwrap(); + let requester = Connection::connect(bus.connect().await.unwrap()).await.unwrap(); + let alias = BusName::new("ai.tinyhumans.Fixture").unwrap(); + original.request_name(alias.clone()).await.unwrap(); + let mut writer = requester.open_stream(&alias, StreamDescriptor::with_len(7)).await.unwrap(); + let reader = original.accept_stream(&writer.stream_ref()).unwrap(); + original.release_name(alias.clone()).await.unwrap(); + replacement.request_name(alias).await.unwrap(); + let received = tokio::spawn(reader.read_to_end_capped(16)); + writer.write(b"payload").await.unwrap(); + writer.finish().await.unwrap(); + assert_eq!(received.await.unwrap().unwrap(), b"payload"); +} From c6438a80d5869c1f0afd084a87033be5c2fc4000 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 11 Oct 2026 19:46:35 +0300 Subject: [PATCH 14/14] fix(connection): pin stream writers to the responder that answered open Stream writers now capture the unique sender from the Open reply and reuse it for every later Write, Close, and Abort, so a stream is not redirected when a well-known name moves to a new owner mid-transfer. Direct raw transports without a stamped sender keep their original destination, and the protocol docs and reply tests were updated to cover the abort path. Auto-committed-on: dragonfly Co-authored-by: Medulla --- crates/tinybus/src/connection.rs | 29 ++++++++++--- crates/tinybus/src/connection_reply_tests.rs | 45 +++++++++++++++++--- docs/protocol.md | 5 ++- 3 files changed, 66 insertions(+), 13 deletions(-) diff --git a/crates/tinybus/src/connection.rs b/crates/tinybus/src/connection.rs index d70fd873..76dc60a9 100644 --- a/crates/tinybus/src/connection.rs +++ b/crates/tinybus/src/connection.rs @@ -986,18 +986,22 @@ impl Connection { descriptor: StreamDescriptor, timeout: Duration, ) -> Result { - let id: String = serde_json::from_value( - self.call_stream_member( + let reply = self + .call_stream_member_message( destination, "Open", serde_json::json!([descriptor]), timeout, ) - .await?, - )?; + .await?; + let id: String = serde_json::from_value(reply.body)?; + // The stream belongs to this responder for its whole lifetime, even + // if the well-known name moves before the next chunk. Direct raw + // transports with no stamped sender retain their original address. + let destination = reply.header.sender.unwrap_or_else(|| destination.clone()); Ok(StreamWriter::new( self.clone(), - destination.clone(), + destination, id, descriptor, timeout, @@ -1175,6 +1179,19 @@ impl Connection { args: Value, timeout: Duration, ) -> Result { + Ok(self + .call_stream_member_message(destination, member, args, timeout) + .await? + .body) + } + + async fn call_stream_member_message( + &self, + destination: &BusName, + member: &str, + args: Value, + timeout: Duration, + ) -> Result { let message = Message::method_call( destination.clone(), ObjectPath::new(STREAM_PATH)?, @@ -1182,7 +1199,7 @@ impl Connection { MemberName::new(member)?, args, ); - self.call_raw(message, timeout).await + self.call_raw_message(message, timeout).await } /// Call a method on the broker's own interface. diff --git a/crates/tinybus/src/connection_reply_tests.rs b/crates/tinybus/src/connection_reply_tests.rs index 40b2d5df..11a9463f 100644 --- a/crates/tinybus/src/connection_reply_tests.rs +++ b/crates/tinybus/src/connection_reply_tests.rs @@ -563,17 +563,50 @@ async fn owner_negotiation_rejects_non_unique_responder_before_delivery() { async fn stream_chunks_and_close_stay_with_the_peer_that_answered_open() { let bus = MemoryBus::new(); Broker::new().spawn(bus.clone()); - let original = Connection::connect(bus.connect().await.unwrap()).await.unwrap(); - let replacement = Connection::connect(bus.connect().await.unwrap()).await.unwrap(); - let requester = Connection::connect(bus.connect().await.unwrap()).await.unwrap(); + let original = Connection::connect(bus.connect().await.unwrap()) + .await + .unwrap(); + let replacement = Connection::connect(bus.connect().await.unwrap()) + .await + .unwrap(); + let requester = Connection::connect(bus.connect().await.unwrap()) + .await + .unwrap(); let alias = BusName::new("ai.tinyhumans.Fixture").unwrap(); original.request_name(alias.clone()).await.unwrap(); - let mut writer = requester.open_stream(&alias, StreamDescriptor::with_len(7)).await.unwrap(); - let reader = original.accept_stream(&writer.stream_ref()).unwrap(); + let mut writer = requester + .open_stream(&alias, StreamDescriptor::with_len(7)) + .await + .unwrap(); + let mut reader = original.accept_stream(&writer.stream_ref()).unwrap(); original.release_name(alias.clone()).await.unwrap(); replacement.request_name(alias).await.unwrap(); - let received = tokio::spawn(reader.read_to_end_capped(16)); + let received = tokio::spawn(async move { reader.read_to_end_capped(16).await }); writer.write(b"payload").await.unwrap(); writer.finish().await.unwrap(); assert_eq!(received.await.unwrap().unwrap(), b"payload"); } + +#[tokio::test] +async fn stream_abort_stays_with_the_peer_that_answered_open() { + let bus = MemoryBus::new(); + Broker::new().spawn(bus.clone()); + let original = Connection::connect(bus.connect().await.unwrap()) + .await + .unwrap(); + let replacement = Connection::connect(bus.connect().await.unwrap()) + .await + .unwrap(); + let requester = Connection::connect(bus.connect().await.unwrap()) + .await + .unwrap(); + let alias = BusName::new("ai.tinyhumans.Fixture").unwrap(); + original.request_name(alias.clone()).await.unwrap(); + let writer = requester + .open_stream(&alias, StreamDescriptor::with_len(7)) + .await + .unwrap(); + original.release_name(alias.clone()).await.unwrap(); + replacement.request_name(alias).await.unwrap(); + writer.abort().await.unwrap(); +} diff --git a/docs/protocol.md b/docs/protocol.md index de2f5973..cf5ec2e9 100644 --- a/docs/protocol.md +++ b/docs/protocol.md @@ -84,7 +84,10 @@ the already selected peer remains the only accepted responder. Old clients omit the optional field and continue routing on a new broker. New clients refuse named calls when an old broker lacks `ResolveCallTarget`; they retain the typed `UnknownMethod` error rather than silently sending an -unguarded call. Unique calls need no alias negotiation. +unguarded call. Unique calls need no alias negotiation. Stream writers retain the unique +responder from `Open` for every later `Write`, `Close`, and `Abort`; they +neither re-resolve an alias per chunk nor redirect an existing stream after +takeover. Unstamped direct-transport replies retain the original destination. These identity guarantees require the actual broker to overwrite ingress senders and protect its transport. A raw `Connection::attach` pair without