diff --git a/crates/tinybus/src/broker.rs b/crates/tinybus/src/broker.rs index 740ac8ea..9eba9da6 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 @@ -341,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"); @@ -614,6 +636,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 5aa246a4..76dc60a9 100644 --- a/crates/tinybus/src/connection.rs +++ b/crates/tinybus/src/connection.rs @@ -141,6 +141,39 @@ 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, + unverified_direct: 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) + || (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. + !brokered + }; + identity_matches + && (reply.header.kind == MessageKind::Error + || !self.confidential + || reply.header.confidential) + } +} + struct Inner { transport: Arc, brokered_native: bool, @@ -150,7 +183,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,9 +745,29 @@ 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; + // 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( @@ -737,16 +790,65 @@ 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 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, tx); + self.inner.pending.lock().await.insert( + serial, + PendingReply { + sender: tx, + expected, + bus_control, + confidential: message.header.confidential, + unverified_direct: self.unique_name().is_none(), + }, + ); 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), @@ -884,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, @@ -1073,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)?, @@ -1080,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. @@ -1206,8 +1325,14 @@ 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 => { @@ -1389,3 +1514,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..11a9463f --- /dev/null +++ b/crates/tinybus/src/connection_reply_tests.rs @@ -0,0 +1,612 @@ +//! Reply authorization regressions over broker-stamped memory deliveries. + +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") + ); +} + +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: &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(), + 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.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.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), + }, + ); + } + 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 = 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(); + 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(), + }, + ) + } 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: InterfaceName::new("ai.tinyhumans.Fixture").unwrap(), member: MemberName::new("Await").unwrap() }.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; +} + +async fn alias_takeover_is_refused(confidential: bool) { + let bus = MemoryBus::new(); + let broker = Broker::new(); + broker.clone().spawn(bus.clone()); + 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(); + #[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, + "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; + 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(); + 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::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()); + 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(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::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(":1.2").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, + unverified_direct: false, + }; + 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, + unverified_direct: 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()); +} + +#[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" + ); +} + +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(_)) + )); +} + +#[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 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(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/crates/tinybus/src/message/mod.rs b/crates/tinybus/src/message/mod.rs index 67355c25..1e3ec670 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,18 @@ impl Message { "only a method call can request a streamed reply", )); } + 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", + )); + } // 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..6e2fddc5 100644 --- a/crates/tinybus/src/router.rs +++ b/crates/tinybus/src/router.rs @@ -483,6 +483,24 @@ 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/docs/protocol.md b/docs/protocol.md index 1645df35..cf5ec2e9 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,35 @@ 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. 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 +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 +185,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` |