bus: handle_requests_with_context — expose the verified caller principal

agent-mesh-bus/src/bus.rs

@@ -28,7 +28,7 @@
 //! (cold-start race) or when the asker never announces at all (a
 //! quiet [`BusOptions`] bind).
 
-use crate::inbox::{BusMessage, Inbox};
+use crate::inbox::{BusMessage, Inbox, RequestContext};
 use crate::reply::CorrelationId;
 use crate::transport::{Inbound, ReplyRoute, Transport};
 use crate::{BusError, Result, Topic};
@@ -325,6 +325,20 @@ impl Bus {
         self.inbox.register_handler(topic, handler);
     }
 
+    /// Register a request handler that also receives the verified caller
+    /// [`RequestContext`] (the authenticated user + agent fingerprints of
+    /// whoever signed the request). Use this when the handler must authorize
+    /// *who* is calling — e.g. a capability-gated responder — rather than serve
+    /// any same-mesh peer. Same synchronous-registration guarantee as
+    /// [`Self::handle_requests`].
+    pub fn handle_requests_with_context<F, Fut>(&self, topic: Topic, handler: F)
+    where
+        F: Fn(RequestContext, Vec<u8>) -> Fut + Send + Sync + 'static,
+        Fut: Future<Output = Result<Vec<u8>>> + Send + 'static,
+    {
+        self.inbox.register_handler_with_context(topic, handler);
+    }
+
     /// Publish a body to `peer_fp` on `topic`. Fire-and-forget — the
     /// caller doesn't wait for a reply. The named peer's bus will
     /// fan it out to any local subscribers on that topic.
@@ -803,6 +817,18 @@ fn spawn_iroh_accept_loop(
     })
 }
 
+/// Whether `env` may be admitted on a QUIC session TLS-authenticated as
+/// `session_id`. The envelope's claimed signer (`cert_chain.agent_pubkey`,
+/// already proven to hold that key by `recv_envelope`'s `env.verify()`) must be
+/// the SAME key that owns the transport session. This binds the application
+/// principal to the session, so a validly-signed envelope replayed or relayed
+/// over a *different* peer's connection is refused rather than authorized as its
+/// original signer. `false` if the claimed pubkey is not a valid ed25519 point
+/// (fail-closed).
+fn envelope_matches_session(session_id: &PublicKey, env: &SignedEnvelope) -> bool {
+    agent_pubkey_to_iroh(&env.cert_chain.agent_pubkey).is_some_and(|signer| &signer == session_id)
+}
+
 /// Handle one accepted connection: finish QUIC, then per bidi stream do the
 /// handshake, decode the envelope, and forward it into `inbound_tx`.
 async fn accept_conn(
@@ -860,6 +886,18 @@ async fn accept_conn(
                 continue;
             }
         };
+        // Bind the principal to the QUIC session (defense in depth over the
+        // signature verify() `recv_envelope` already did): the envelope's signer
+        // must be the key that TLS-authenticated THIS connection, so a
+        // validly-signed envelope relayed/replayed over another peer's session
+        // is dropped here instead of being authorized as its original signer.
+        if !envelope_matches_session(&conn.remote_id(), &env) {
+            tracing::warn!(
+                signer = %env.sender_agent_fp().short(),
+                "iroh transport: envelope signer is not bound to the QUIC session identity; dropping"
+            );
+            continue;
+        }
         if inbound_tx
             .send(Inbound {
                 envelope: env,
@@ -893,6 +931,34 @@ mod tests {
         )
     }
 
+    /// The QUIC-session binding: an envelope is admitted only on a session
+    /// authenticated as its own signer. A validly-signed envelope presented over
+    /// a *sibling's* session (relay/replay) is refused — the principal is bound
+    /// to the transport session, not only to the envelope signature. Pure
+    /// regression for the `accept_conn` session-binding hardening (newt#1643 /
+    /// agent-mesh#75 follow-up).
+    #[test]
+    fn an_envelope_is_bound_to_its_signers_quic_session() {
+        let user = UserKey::generate();
+        let a = agent(&user, "a");
+        let b = agent(&user, "b");
+        let a_session = agent_pubkey_to_iroh(&a.public_bytes()).expect("valid ed25519 key");
+        let b_session = agent_pubkey_to_iroh(&b.public_bytes()).expect("valid ed25519 key");
+        let env = SignedEnvelope::new(
+            &a,
+            Recipient::Direct {
+                agent_fp: Fingerprint::of_bytes(&b.public_bytes()),
+            },
+            1,
+            b"payload".to_vec(),
+        );
+        // Admitted on A's own session (the signer owns the transport)…
+        assert!(envelope_matches_session(&a_session, &env));
+        // …refused on B's session — a relayed/replayed envelope can't borrow B's
+        // connection to speak as A.
+        assert!(!envelope_matches_session(&b_session, &env));
+    }
+
     /// The request/reply round-trip driven over the **in-memory
     /// transport** — the exact same `Bus` send / receive / inbox / reply
     /// wiring the iroh path uses, but with no sockets, no mDNS, and no QUIC
@@ -937,6 +1003,47 @@ mod tests {
         bob_bus.close().await.unwrap();
     }
 
+    #[tokio::test]
+    async fn context_handler_sees_the_calling_agent_over_the_transport() {
+        let user = UserKey::generate();
+        let alice = Arc::new(agent(&user, "alice"));
+        let bob = Arc::new(agent(&user, "bob"));
+        let alice_fp = alice.fingerprint();
+        let bob_fp = bob.fingerprint();
+
+        let net = MeshNet::new();
+        let alice_bus = Bus::bind_with_transport(
+            alice,
+            user.fingerprint(),
+            Arc::new(net.transport_for(alice_fp)),
+        );
+        let bob_bus =
+            Bus::bind_with_transport(bob, user.fingerprint(), Arc::new(net.transport_for(bob_fp)));
+
+        let topic = Topic::new(user.fingerprint(), "whoami");
+        bob_bus.handle_requests_with_context(
+            topic.clone(),
+            |ctx: RequestContext, _body| async move {
+                // The responder learns WHO called from the verified envelope, not
+                // from anything the caller put in the body.
+                Ok(ctx.caller_agent_fp.hex().into_bytes())
+            },
+        );
+
+        let reply = alice_bus
+            .request(bob_fp, &topic, b"".to_vec(), Duration::from_secs(5))
+            .await
+            .expect("round-trip reply");
+        assert_eq!(
+            String::from_utf8(reply).unwrap(),
+            alice_fp.hex(),
+            "the responder must see ALICE as the caller"
+        );
+
+        alice_bus.close().await.unwrap();
+        bob_bus.close().await.unwrap();
+    }
+
     /// Regression (#52 de-flake): `handle_requests` must register the
     /// handler *before it returns*, with no spawn and no intervening
     /// yield. It used to spawn the registration onto the runtime, so on a

agent-mesh-bus/src/inbox.rs

@@ -74,10 +74,34 @@ pub enum BusMessage {
     },
 }
 
-/// Type of a registered request handler. Takes the request body,
-/// returns the reply body asynchronously.
-pub type RequestHandler =
-    Arc<dyn Fn(Vec<u8>) -> BoxFuture<'static, Result<Vec<u8>>> + Send + Sync + 'static>;
+/// The verified principal behind an inbound request: who signed the envelope.
+///
+/// Both fingerprints are authenticated, not claimed. Every inbound envelope is
+/// `verify()`-ed at the transport boundary (`recv_envelope`) before it reaches
+/// the inbox — the agent signature is checked against the envelope's cert chain,
+/// and the chain proves the user→agent delegation. So `caller_agent_fp` is
+/// `BLAKE3(cert_chain.agent_pubkey)` of whoever actually signed this request,
+/// and `caller_user_fp` is their operator root. A handler may authorize on these
+/// without re-verifying anything.
+///
+/// This is the *signer* of the request, which is the correct principal for
+/// authorization: a relay can only deliver a request its signer already
+/// authorized, never mint one under another agent's key.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct RequestContext {
+    /// The caller's operator root fingerprint (`env.sender_user_fp()`).
+    pub caller_user_fp: Fingerprint,
+    /// The caller's agent fingerprint (`env.sender_agent_fp()` =
+    /// `BLAKE3(agent_pubkey)`), the handle a capability registry keys on.
+    pub caller_agent_fp: Fingerprint,
+}
+
+/// Type of a registered request handler. Takes the verified caller
+/// [`RequestContext`] and the request body, returns the reply body
+/// asynchronously.
+pub type RequestHandler = Arc<
+    dyn Fn(RequestContext, Vec<u8>) -> BoxFuture<'static, Result<Vec<u8>>> + Send + Sync + 'static,
+>;
 
 /// What the bus should send out in response to an incoming envelope.
 ///
@@ -157,9 +181,29 @@ impl Inbox {
     where
         F: Fn(Vec<u8>) -> Fut + Send + Sync + 'static,
         Fut: Future<Output = Result<Vec<u8>>> + Send + 'static,
+    {
+        // The context-free convenience: discard the caller principal. Kept so
+        // existing body-only handlers need no change.
+        let key = topic.wire();
+        let boxed: RequestHandler = Arc::new(move |_ctx, body| Box::pin(handler(body)));
+        self.handlers
+            .write()
+            .expect("handlers lock poisoned")
+            .insert(key, boxed);
+    }
+
+    /// Register a request handler that receives the verified [`RequestContext`]
+    /// (the caller's authenticated user + agent fingerprints) alongside the
+    /// body — for handlers that must authorize *who* is calling, not just serve
+    /// the request. Same synchronous-registration guarantee as
+    /// [`Self::register_handler`].
+    pub fn register_handler_with_context<F, Fut>(&self, topic: Topic, handler: F)
+    where
+        F: Fn(RequestContext, Vec<u8>) -> Fut + Send + Sync + 'static,
+        Fut: Future<Output = Result<Vec<u8>>> + Send + 'static,
     {
         let key = topic.wire();
-        let boxed: RequestHandler = Arc::new(move |body| Box::pin(handler(body)));
+        let boxed: RequestHandler = Arc::new(move |ctx, body| Box::pin(handler(ctx, body)));
         self.handlers
             .write()
             .expect("handlers lock poisoned")
@@ -232,16 +276,21 @@ impl Inbox {
             });
         }
 
+        // Build the verified caller principal from the (already-verified)
+        // envelope. Both fingerprints are authenticated by env.verify() at the
+        // transport boundary — see RequestContext.
+        let ctx = RequestContext {
+            caller_user_fp: env.sender_user_fp(),
+            caller_agent_fp: peer_fp,
+        };
+
         let msg: BusMessage = serde_json::from_slice(env.payload.as_ref())?;
         match msg {
             BusMessage::Request {
                 topic,
                 correlation,
                 body,
-            } => {
-                self.dispatch_request(peer_fp, topic, correlation, body)
-                    .await
-            }
+            } => self.dispatch_request(ctx, topic, correlation, body).await,
             BusMessage::Reply { correlation, body } => {
                 let cid = CorrelationId(correlation);
                 let delivered = self.waiters.deliver(cid, body);
@@ -262,7 +311,7 @@ impl Inbox {
 
     async fn dispatch_request(
         &self,
-        peer_fp: Fingerprint,
+        ctx: RequestContext,
         topic: String,
         correlation: [u8; 16],
         body: Vec<u8>,
@@ -275,7 +324,8 @@ impl Inbox {
             tracing::debug!(topic = %topic, "inbox: no handler for request topic");
             return Ok(None);
         };
-        let reply_body = handler(body).await?;
+        let peer_fp = ctx.caller_agent_fp;
+        let reply_body = handler(ctx, body).await?;
         Ok(Some(OutgoingReply {
             peer_fp,
             correlation: CorrelationId(correlation),
@@ -440,6 +490,45 @@ mod tests {
         assert_eq!(out.body, b"echo:hi");
     }
 
+    #[tokio::test]
+    async fn a_context_handler_receives_the_verified_caller_fingerprints() {
+        let user = UserKey::generate();
+        let alice = agent(&user, "alice");
+        let bob_fp = agent(&user, "bob").fingerprint();
+        let topic = Topic::new(user.fingerprint(), "whoami");
+
+        let inbox = Inbox::new();
+        // The handler echoes back the caller principal it was handed, so the
+        // test can prove it is the ACTUAL signer of the envelope (alice), not a
+        // value copied from the request body.
+        inbox.register_handler_with_context(
+            topic.clone(),
+            |ctx: RequestContext, _body| async move {
+                Ok(
+                    format!("{}|{}", ctx.caller_user_fp.hex(), ctx.caller_agent_fp.hex())
+                        .into_bytes(),
+                )
+            },
+        );
+
+        let req = BusMessage::Request {
+            topic: topic.wire(),
+            correlation: [0x7; 16],
+            body: b"ignored".to_vec(),
+        };
+        let out = inbox
+            .on_envelope(envelope(&alice, bob_fp, 1, &req))
+            .await
+            .unwrap()
+            .expect("reply produced");
+        let got = String::from_utf8(out.body).unwrap();
+        assert_eq!(
+            got,
+            format!("{}|{}", user.fingerprint().hex(), alice.fingerprint().hex()),
+            "the handler must see alice's authenticated user+agent fingerprints"
+        );
+    }
+
     #[tokio::test]
     async fn request_with_no_handler_returns_none() {
         let user = UserKey::generate();

agent-mesh-bus/src/lib.rs

@@ -28,7 +28,7 @@ pub mod pyo3_module;
 
 pub use bus::{Bus, BusOptions, IrohTransport, PeerEndpoint};
 pub use error::{BusError, Result};
-pub use inbox::{BusMessage, Inbox, OutgoingReply};
+pub use inbox::{BusMessage, Inbox, OutgoingReply, RequestContext};
 pub use reply::CorrelationId;
 pub use topic::Topic;
 pub use transport::{InMemoryTransport, Inbound, MeshNet, ReplyRoute, Transport};