diff --git a/agent-mesh-bus/src/bus.rs b/agent-mesh-bus/src/bus.rs index c9cf5ad..a8bb518 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. @@ -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 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};