Bind replies to the expected responder and guard alias routing

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<Value> {
+                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"),
                 ),

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<Message>,
+    expected: Option<BusName>,
+    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<dyn Transport>,
     brokered_native: bool,
@@ -150,7 +183,7 @@ struct Inner {
     /// concurrently without interleaving two frames on the wire.
     outbox: mpsc::Sender<Message>,
     serial: AtomicU64,
-    pending: Mutex<HashMap<u64, oneshot::Sender<Message>>>,
+    pending: Mutex<HashMap<u64, PendingReply>>,
     objects: RwLock<ObjectTree>,
     /// 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<Value> {
+        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<Message> {
-        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<StreamWriter> {
-        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,14 +1179,27 @@ impl Connection {
         args: Value,
         timeout: Duration,
     ) -> Result<Value> {
+        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<Message> {
         let message = Message::method_call(
             destination.clone(),
             ObjectPath::new(STREAM_PATH)?,
             InterfaceName::new(STREAM_INTERFACE)?,
             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<Inner>) {
                 };
                 // 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;