From 601dca2533d4abc871705ea2effe18b4ffa55af3 Mon Sep 17 00:00:00 2001 From: hartsock Date: Tue, 11 Aug 2026 22:57:20 -0400 Subject: [PATCH 1/2] =?UTF-8?q?feat(bus):=20handle=5Frequests=5Fwith=5Fcon?= =?UTF-8?q?text=20=E2=80=94=20expose=20the=20verified=20caller=20principal?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Request handlers registered via `Bus::handle_requests` receive only the request body: the authenticated caller identity, though known, is discarded. That is fine for peers that serve any same-mesh caller, but a capability-gated responder (e.g. a dock service that must authorize WHICH agent is calling) cannot express its policy — authorization is forced onto the dialer, which is not complete mediation. Add an additive, wire-compatible seam: - `RequestContext { caller_user_fp, caller_agent_fp }` — the verified principal. Every inbound envelope is already `verify()`-ed at the transport boundary, so `sender_agent_fp()` / `sender_user_fp()` are authenticated (not claimed); this is the request's *signer*, the correct principal for authorization (a relay can only deliver a request its signer already authorized). - `Inbox::register_handler_with_context` + `Bus::handle_requests_with_context`, which hand the handler `(RequestContext, body)`. The existing body-only `handle_requests` / `register_handler` are unchanged (they wrap, discarding the context), so no caller churns. No transport, wire, or protocol change: the caller fingerprints are already computed in `on_envelope`; this threads them through `dispatch_request` to the handler. ## What this PR does - add `RequestContext` and the `*_with_context` handler registration to the bus - keep the body-only API as a zero-context wrapper (back-compat) ## Test plan - `inbox::a_context_handler_receives_the_verified_caller_fingerprints` — the handler sees the ACTUAL envelope signer's user+agent fingerprints, not a body value - `bus::context_handler_sees_the_calling_agent_over_the_transport` — full in-memory round-trip proving the responder learns the caller from the verified envelope - `cargo test -p agent-mesh-bus` 59 pass; `clippy --all-targets -D warnings` clean ## Out of scope - binding the principal to the QUIC session key (`conn.remote_id()`) as an additional cross-check — the envelope signature already authenticates the signer; the connection cross-check is defense-in-depth, a follow-up. Co-Authored-By: Claude Fable 5 --- agent-mesh-bus/src/bus.rs | 57 +++++++++++++++++- agent-mesh-bus/src/inbox.rs | 111 ++++++++++++++++++++++++++++++++---- agent-mesh-bus/src/lib.rs | 2 +- 3 files changed, 157 insertions(+), 13 deletions(-) diff --git a/agent-mesh-bus/src/bus.rs b/agent-mesh-bus/src/bus.rs index c9cf5ad..d3892fd 100644 --- a/agent-mesh-bus/src/bus.rs +++ b/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(&self, topic: Topic, handler: F) + where + F: Fn(RequestContext, Vec) -> Fut + Send + Sync + 'static, + Fut: Future>> + 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. @@ -937,6 +951,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 diff --git a/agent-mesh-bus/src/inbox.rs b/agent-mesh-bus/src/inbox.rs index 5551182..d61db82 100644 --- a/agent-mesh-bus/src/inbox.rs +++ b/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) -> BoxFuture<'static, Result>> + 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) -> BoxFuture<'static, Result>> + 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) -> Fut + Send + Sync + 'static, Fut: Future>> + 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(&self, topic: Topic, handler: F) + where + F: Fn(RequestContext, Vec) -> Fut + Send + Sync + 'static, + Fut: Future>> + 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, @@ -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(); diff --git a/agent-mesh-bus/src/lib.rs b/agent-mesh-bus/src/lib.rs index 1c01ee9..4905ac0 100644 --- a/agent-mesh-bus/src/lib.rs +++ b/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}; From 8e4d60861f7524d1aca1ff4fd85e855b409d254e Mon Sep 17 00:00:00 2001 From: Shawn Hartsock Date: Wed, 12 Aug 2026 07:59:15 -0400 Subject: [PATCH 2/2] feat(bus): bind the request principal to the QUIC session in accept_conn recv_envelope already verifies the envelope signature, but accept_conn forwarded the envelope without checking that its signer owns the QUIC session it arrived on. A validly-signed envelope replayed or relayed over a different peer's connection would then be authorized as its original signer. Add a pure envelope_matches_session(session_id, env) helper and reject at accept_conn any envelope whose cert_chain.agent_pubkey does not match conn.remote_id(). This binds the application principal to the transport session (defense in depth over the signature verify). Proven by an_envelope_is_bound_to_its_signers_quic_session and, end-to-end, by the newt-mesh live loopback-QUIC dock tests (approved hub still served, unapproved sibling still denied). Follow-up noted in #75. Co-Authored-By: Claude Opus 5 --- agent-mesh-bus/src/bus.rs | 52 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 52 insertions(+) diff --git a/agent-mesh-bus/src/bus.rs b/agent-mesh-bus/src/bus.rs index d3892fd..a8bb518 100644 --- a/agent-mesh-bus/src/bus.rs +++ b/agent-mesh-bus/src/bus.rs @@ -817,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( @@ -874,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, @@ -907,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