From f81a9bf93f9d83edb5b69a2d87d121ca44907c65 Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Thu, 17 Sep 2026 22:12:45 +0100 Subject: [PATCH 1/2] feat(core): add a tokio-based session runtime over the Connection port Adds Session, the Rust node/session runtime wire-mesh#173 identified as missing: wraps an already-established Connection, exchanges and negotiates the handshake, dispatches incoming manage-requests to handlers registered per capability verb via a new HandlerRegistry, resolves send_manage_request calls against their correlated manage-response (with an optional timeout resolving a protocol-level timeout outcome rather than hanging forever), sends data frames directly, and emits connection lifecycle events over an unbounded channel. Uses tokio tasks and channels rather than TypeScript's async-iterable mesh-session.ts shape, per the issue's own design decision: a session runtime is a concurrency-model-specific construct, unlike the byte-level codec or synchronous verification-logic parity every other Rust/TypeScript pairing in this codebase has. Gated behind the existing "net" feature, alongside tcp_transport.rs, since it needs a real tokio task-spawning runtime that wasm32-unknown-unknown consumers (wire-mesh-threshold-wasm) cannot support. --- rust/crates/wire-mesh-core/src/domain/mod.rs | 6 + .../wire-mesh-core/src/domain/session.rs | 708 ++++++++++++++++++ 2 files changed, 714 insertions(+) create mode 100644 rust/crates/wire-mesh-core/src/domain/session.rs diff --git a/rust/crates/wire-mesh-core/src/domain/mod.rs b/rust/crates/wire-mesh-core/src/domain/mod.rs index ea9067a..1d612eb 100644 --- a/rust/crates/wire-mesh-core/src/domain/mod.rs +++ b/rust/crates/wire-mesh-core/src/domain/mod.rs @@ -9,6 +9,8 @@ pub mod revocation; pub mod room; pub mod room_path; pub mod room_token_verification; +#[cfg(feature = "net")] +pub mod session; pub mod shard_manifest; pub mod tokens; @@ -19,4 +21,8 @@ pub use room_path::{device_id_from_hex, device_id_to_hex, parse_room_path, Parse pub use room_token_verification::{ verify_room_token, RoomTokenRejection, RoomTokenVerdict, ROOM_MEMBER_CAPABILITY, }; +#[cfg(feature = "net")] +pub use session::{ + DataFrame, HandlerRegistry, IncomingManageRequest, ManageRequestHandler, Session, SessionEvent, +}; pub use tokens::{verify_capability_token, TokenRejection, TokenVerdict}; diff --git a/rust/crates/wire-mesh-core/src/domain/session.rs b/rust/crates/wire-mesh-core/src/domain/session.rs new file mode 100644 index 0000000..793971f --- /dev/null +++ b/rust/crates/wire-mesh-core/src/domain/session.rs @@ -0,0 +1,708 @@ +//! The Rust node/session runtime (wire-mesh#173): the concurrent +//! counterpart to the TypeScript core's `mesh-session.ts`, built on +//! Rust's own idiom for a stateful, event-driven dispatch loop -- tasks +//! and channels -- rather than a port of `mesh-session.ts`'s +//! async-iterable shape. See wire-mesh#173's own issue body for the +//! design decision and why the two sides deliberately diverge in API +//! shape while matching protocol behaviour: same frames sent and +//! received, same state machine. +//! +//! Scope, matching the issue: wrap an already-established [`Connection`] +//! (dialled or accepted -- the post-connect protocol handshake is +//! symmetric either way, so this has no dial/reconnect logic of its own), +//! perform the handshake exchange, dispatch incoming manage-requests to +//! handlers registered per capability verb, emit connection lifecycle +//! events, and support sending a manage-request (awaiting its correlated +//! response, with an optional timeout) and sending a data frame. Gossip +//! directory, relay pairing, and revocation-announcement handling are +//! deliberately out of scope -- #171's threshold coordinator/participant +//! handlers, this issue's own stated consumer, need none of them. + +use core::future::poll_fn; +use core::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use core::time::Duration; +use std::collections::HashMap; +use std::pin::Pin; +use std::sync::{Arc, Mutex as StdMutex}; + +use futures_core::Stream; +use tokio::sync::{mpsc, oneshot, Mutex as AsyncMutex}; +use tokio::task::AbortHandle; +use wire_mesh_wire::handshake::{DomainId, HandshakeFrame, ProtocolVersion}; +use wire_mesh_wire::management::{ + ManageCommand, ManageError, ManageOutcome, ManageRequestFrame, ManageResponseFrame, +}; +use wire_mesh_wire::tokens::{CapabilityScope, CoseSign1}; +use wire_mesh_wire::Frame; + +use crate::domain::handshake::{negotiate, SUPPORTED_PROTOCOL_VERSION}; +use crate::ports::{Connection, CoreError}; + +/// A connection shared between the dispatch loop (which reads frames) +/// and every caller of `send_manage_request`/`send_data_frame` (which +/// write them) -- `send`/`close` only need shared, not exclusive, logical +/// ownership of the connection's *lifetime*, but `Connection::close` is +/// `&mut self`, so a lock is the simplest correct way to share one +/// `Box` across tasks without unsafe aliasing. +type SharedConnection = Arc>>; + +/// A caller awaiting one `send_manage_request`'s correlated response, or +/// being told the connection ended (or was closed locally) before one +/// arrived -- see `send_manage_request`'s own doc comment for why this is +/// `Result` rather than always resolving an outcome: a disconnect is a +/// transport-level failure (surfaced as `Err`), while a timeout is a +/// protocol-level, expected outcome (surfaced as +/// `Ok(ManageOutcome::Error { code: "timeout", .. })`), mirroring +/// `mesh-session.ts`'s own reject-on-disconnect/resolve-on-timeout split. +type PendingReply = Result; +type PendingMap = Arc>>>; + +/// One manage-request this session received from its peer, handed to +/// whichever [`ManageRequestHandler`] is registered for its command's +/// verb. +#[derive(Debug)] +pub struct IncomingManageRequest { + pub request_id: u64, + pub command: ManageCommand, + pub scope: CapabilityScope, + pub token: Option, +} + +/// A handler for every manage-request whose command carries one specific +/// capability verb, registered against a [`HandlerRegistry`]. Returning +/// an outcome directly (rather than TS's callback-style `respond()`) is +/// possible because Rust handlers are registered up front rather than +/// pulled one at a time off an iterator -- the real registration +/// mechanism `mesh-session.ts` itself never had, and the gap wire-mesh#173 +/// exists to close. +#[async_trait::async_trait] +pub trait ManageRequestHandler: Send + Sync { + async fn handle(&self, request: IncomingManageRequest) -> ManageOutcome; +} + +/// Maps a capability verb (`ManageCommand.verb`, e.g. +/// `"exadev.io/threshold:sign"`) to the handler that owns every command +/// shape carried under it. A manage-request whose verb has no registered +/// handler gets an explicit `manage-error { code: "unknown-verb" }` +/// response -- loud and immediate, never a silently-dropped frame the +/// sender is left waiting forever for. +#[derive(Default)] +pub struct HandlerRegistry { + handlers: HashMap>, +} + +impl HandlerRegistry { + pub fn new() -> Self { + Self::default() + } + + /// Registers `handler` for `verb`, replacing whatever was previously + /// registered for that same verb. Consuming/returning `Self` gives a + /// builder-style call chain (`HandlerRegistry::new().register(...).register(...)`). + #[must_use] + pub fn register( + mut self, + verb: impl Into, + handler: Arc, + ) -> Self { + self.handlers.insert(verb.into(), handler); + self + } + + fn get(&self, verb: &str) -> Option> { + self.handlers.get(verb).cloned() + } +} + +/// A `core/data` frame `send_data_frame` can send directly over a +/// session's own connection -- the transport half of an application's +/// own noticeboard replication policy, mirroring +/// `mesh-session.ts`'s own `sendDataFrame` union parameter. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum DataFrame { + Have(wire_mesh_wire::data::DataHaveFrame), + Request(wire_mesh_wire::data::DataRequestFrame), + Entries(wire_mesh_wire::data::DataEntriesFrame), +} + +impl DataFrame { + fn into_frame(self) -> Frame { + match self { + DataFrame::Have(f) => Frame::DataHave(f), + DataFrame::Request(f) => Frame::DataRequest(f), + DataFrame::Entries(f) => Frame::DataEntries(f), + } + } +} + +/// A connection/session lifecycle event, delivered on the receiver +/// `Session::accept` hands back alongside the `Session` itself. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum SessionEvent { + /// The peer's handshake negotiated successfully. + HandshakeNegotiated { + version: u64, + shared_domains: Vec, + }, + /// The peer's handshake was received but negotiation failed (version + /// this implementation cannot support, or an invalid domain on + /// either side). + HandshakeRejected { reason: String }, + /// The connection ended, whether the peer closed it, the underlying + /// transport failed, or `Session::close` was called locally. + Disconnected { reason: String }, +} + +async fn next_frame(stream: &mut Pin + Send>>) -> Option { + poll_fn(|cx| stream.as_mut().poll_next(cx)).await +} + +async fn send_frame(connection: &SharedConnection, frame: Frame) -> Result<(), CoreError> { + connection.lock().await.send(frame).await +} + +fn reject_all_pending(pending: &PendingMap, reason: &str) { + let mut senders: Vec> = { + let mut map = pending.lock().unwrap_or_else(|e| e.into_inner()); + map.drain().map(|(_, sender)| sender).collect() + }; + for sender in senders.drain(..) { + let _ = sender.send(Err(reason.to_owned())); + } +} + +/// Drives the dispatch loop for as long as the connection's inbound +/// stream yields frames: negotiates the handshake (once -- a second +/// handshake frame is ignored, mirroring `mesh-session.ts`'s own +/// `if (handshake.status !== "pending") return`), resolves pending +/// `send_manage_request` calls against a matching `manage-response`, and +/// spawns one task per incoming `manage-request` so a slow handler never +/// blocks reading further frames off the same connection. Frame kinds +/// outside this runtime's stated scope (gossip, relay, revocation, +/// ping/close) are read (so the stream keeps advancing) but otherwise +/// ignored. +#[allow(clippy::too_many_arguments)] +async fn run_dispatch_loop( + mut inbound: Pin + Send>>, + connection: SharedConnection, + pending: PendingMap, + registry: Arc, + events_tx: mpsc::UnboundedSender, + local_handshake: HandshakeFrame, + closed: Arc, +) { + let mut handshake_settled = false; + while let Some(frame) = next_frame(&mut inbound).await { + match frame { + Frame::Handshake(remote) => { + if handshake_settled { + continue; + } + handshake_settled = true; + match negotiate(&local_handshake, &remote) { + Ok(result) => { + let _ = events_tx.send(SessionEvent::HandshakeNegotiated { + version: result.version, + shared_domains: result.shared_domains, + }); + } + Err(error) => { + let _ = events_tx.send(SessionEvent::HandshakeRejected { + reason: error.to_string(), + }); + } + } + } + Frame::ManageResponse(ManageResponseFrame { + request_id, + outcome, + }) => { + let sender = { + let mut map = pending.lock().unwrap_or_else(|e| e.into_inner()); + map.remove(&request_id) + }; + if let Some(sender) = sender { + let _ = sender.send(Ok(outcome)); + } + } + Frame::ManageRequest(request) => { + let ManageRequestFrame { + request_id, + command, + scope, + token, + } = *request; + let registry = Arc::clone(®istry); + let connection = Arc::clone(&connection); + tokio::spawn(async move { + let verb = command.verb.0.clone(); + let handler = registry.get(&verb); + let outcome = match handler { + Some(handler) => { + handler + .handle(IncomingManageRequest { + request_id, + command, + scope, + token, + }) + .await + } + None => ManageOutcome::Error(ManageError { + code: "unknown-verb".to_owned(), + message: Some(verb), + }), + }; + let response = Frame::ManageResponse(ManageResponseFrame { + request_id, + outcome, + }); + let _ = send_frame(&connection, response).await; + }); + } + // Out of scope for this runtime (gossip directory, relay + // pairing, revocation announcements, ping/close housekeeping) + // -- observed so the stream keeps advancing, never acted on. + _ => {} + } + } + closed.store(true, Ordering::SeqCst); + reject_all_pending(&pending, "connection closed before a response arrived"); + let _ = events_tx.send(SessionEvent::Disconnected { + reason: "the peer's connection ended".to_owned(), + }); +} + +/// A live session over one [`Connection`]: the runtime this crate +/// otherwise lacked (wire-mesh#173). Built by [`Session::accept`], which +/// works identically whether the connection was dialled +/// (`Transport::connect`) or accepted (a `Transport::listen` callback) -- +/// the post-connect protocol handshake is symmetric either way, and this +/// runtime has no dial/reconnect logic of its own to distinguish the two. +pub struct Session { + connection: SharedConnection, + next_request_id: AtomicU64, + pending: PendingMap, + dispatch_abort: AbortHandle, + events_tx: mpsc::UnboundedSender, + closed: Arc, +} + +impl Session { + /// Wires up an already-established connection: sends this side's own + /// handshake, then starts the dispatch loop (spawned as a background + /// task) that negotiates the peer's answering handshake, dispatches + /// incoming manage-requests to `registry`, and resolves outstanding + /// `send_manage_request` calls. Returns the session together with + /// the receiver its lifecycle events arrive on. + pub async fn accept( + connection: Box, + local_domains: impl IntoIterator, + registry: HandlerRegistry, + ) -> Result<(Session, mpsc::UnboundedReceiver), CoreError> { + let mut connection = connection; + let inbound = connection.receive()?; + let shared: SharedConnection = Arc::new(AsyncMutex::new(connection)); + + let local_handshake = HandshakeFrame { + version: ProtocolVersion(SUPPORTED_PROTOCOL_VERSION), + domains: local_domains.into_iter().map(DomainId).collect(), + params: None, + }; + send_frame(&shared, Frame::Handshake(local_handshake.clone())).await?; + + let pending: PendingMap = Arc::new(StdMutex::new(HashMap::new())); + let (events_tx, events_rx) = mpsc::unbounded_channel(); + let closed = Arc::new(AtomicBool::new(false)); + + let task = tokio::spawn(run_dispatch_loop( + inbound, + Arc::clone(&shared), + Arc::clone(&pending), + Arc::new(registry), + events_tx.clone(), + local_handshake, + Arc::clone(&closed), + )); + + Ok(( + Session { + connection: shared, + next_request_id: AtomicU64::new(0), + pending, + dispatch_abort: task.abort_handle(), + events_tx, + closed, + }, + events_rx, + )) + } + + /// Sends a manage-request and resolves with the matching + /// manage-response's outcome, correlated by request-id. When + /// `timeout` is given, elapsing it resolves + /// `Ok(ManageOutcome::Error { code: "timeout", .. })` rather than + /// waiting forever, mirroring `sendManageRequest`'s own `timeoutMs`. + /// A connection that ends (peer disconnect, or a local + /// [`Session::close`]) before a response arrives resolves `Err`, not + /// an outcome -- unlike a timeout, a lost connection is not an + /// ordinary protocol-level answer the peer chose to give. + pub async fn send_manage_request( + &self, + command: ManageCommand, + scope: CapabilityScope, + token: Option, + timeout: Option, + ) -> Result { + if self.closed.load(Ordering::SeqCst) { + return Err(CoreError::Transport("session is closed".to_owned())); + } + let request_id = self.next_request_id.fetch_add(1, Ordering::Relaxed); + let (tx, rx) = oneshot::channel::(); + { + let mut map = self.pending.lock().unwrap_or_else(|e| e.into_inner()); + map.insert(request_id, tx); + } + let frame = Frame::ManageRequest(Box::new(ManageRequestFrame { + request_id, + command, + scope, + token, + })); + if let Err(error) = send_frame(&self.connection, frame).await { + let mut map = self.pending.lock().unwrap_or_else(|e| e.into_inner()); + map.remove(&request_id); + return Err(error); + } + + let outcome = match timeout { + None => rx.await, + Some(duration) => match tokio::time::timeout(duration, rx).await { + Ok(result) => result, + Err(_elapsed) => { + let mut map = self.pending.lock().unwrap_or_else(|e| e.into_inner()); + map.remove(&request_id); + return Ok(ManageOutcome::Error(ManageError { + code: "timeout".to_owned(), + message: None, + })); + } + }, + }; + + match outcome { + Ok(Ok(outcome)) => Ok(outcome), + Ok(Err(reason)) => Err(CoreError::Transport(reason)), + // The sender was dropped without sending -- this runtime + // always sends a reply on disconnect (see + // `reject_all_pending`) and on a resolved response, so this + // arm is only reachable if the dispatch task itself panicked. + Err(_recv_error) => Err(CoreError::Transport( + "connection closed before a response arrived".to_owned(), + )), + } + } + + /// Sends one `core/data` frame directly over this session's own + /// connection. + pub async fn send_data_frame(&self, frame: DataFrame) -> Result<(), CoreError> { + if self.closed.load(Ordering::SeqCst) { + return Err(CoreError::Transport("session is closed".to_owned())); + } + send_frame(&self.connection, frame.into_frame()).await + } + + /// Closes the underlying connection, rejects every + /// `send_manage_request` call still awaiting a response, and stops + /// the dispatch loop. Idempotent: closing an already-closed session + /// is a no-op. + pub async fn close(&self) -> Result<(), CoreError> { + if self.closed.swap(true, Ordering::SeqCst) { + return Ok(()); + } + self.dispatch_abort.abort(); + reject_all_pending(&self.pending, "connection closed before a response arrived"); + let _ = self.events_tx.send(SessionEvent::Disconnected { + reason: "closed locally".to_owned(), + }); + self.connection.lock().await.close().await + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::adapters::TcpTransport; + use crate::ports::{OnConnection, Transport}; + use std::net::SocketAddr; + use tokio::sync::mpsc as tokio_mpsc; + use wire_mesh_wire::data::DataHaveFrame; + use wire_mesh_wire::identity::DeviceId; + use wire_mesh_wire::management::ManageOk; + use wire_mesh_wire::tokens::CapabilityVerb; + use wire_mesh_wire::value::CanonicalMap; + + /// Binds a fresh TCP loopback pair -- one side accepted via + /// `Transport::listen`, the other dialled via `Transport::connect` -- + /// so every test below exercises the runtime over a real connection, + /// never a hand-rolled test double. + async fn tcp_pair() -> (Box, Box) { + let transport = TcpTransport::new(); + let (accepted_tx, mut accepted_rx) = tokio_mpsc::unbounded_channel::>(); + let on_connection: OnConnection = Arc::new(move |connection| { + let _ = accepted_tx.send(connection); + }); + let guard = transport + .listen("127.0.0.1:0", on_connection) + .await + .expect("listen"); + let bound = guard + .address() + .parse::() + .expect("tcp reports a parseable host:port address"); + let dialled = transport + .connect(&format!("127.0.0.1:{}", bound.port())) + .await + .expect("connect"); + let accepted = accepted_rx.recv().await.expect("accepted connection"); + (dialled, accepted) + } + + fn scope() -> CapabilityScope { + CapabilityScope { + kind: "folder".to_owned(), + path: None, + } + } + + struct EchoHandler; + + #[async_trait::async_trait] + impl ManageRequestHandler for EchoHandler { + async fn handle(&self, request: IncomingManageRequest) -> ManageOutcome { + let mut extra = CanonicalMap::new(); + extra + .insert( + "echoed-request-id".to_owned(), + wire_mesh_wire::value::CborValue::UInt(request.request_id), + ) + .expect("unique key"); + ManageOutcome::Ok(ManageOk { extra }) + } + } + + /// Builds a `ManageCommand` whose params are the open `Json` catch-all + /// -- which itself requires an inner `"verb"` key (`manage_params_from` + /// looks for one to pick a decode arm, and the closed arms all fail to + /// match an arbitrary test verb), distinct from `command.verb` itself, + /// the outer *capability* verb this session dispatches handlers by. + fn json_command(verb: &str) -> ManageCommand { + let mut params = CanonicalMap::new(); + params + .insert( + "verb".to_owned(), + wire_mesh_wire::value::CborValue::Text(verb.to_owned()), + ) + .expect("unique key"); + ManageCommand { + verb: CapabilityVerb(verb.to_owned()), + params: wire_mesh_wire::management::ManageParams::Json(params), + } + } + + #[tokio::test] + async fn accepting_both_sides_negotiates_the_shared_domain() { + let (a, b) = tcp_pair().await; + let (session_a, mut events_a) = Session::accept( + a, + ["core/management".to_owned(), "core/exec".to_owned()], + HandlerRegistry::new(), + ) + .await + .expect("accept a"); + let (session_b, mut events_b) = Session::accept( + b, + ["core/exec".to_owned(), "core/data".to_owned()], + HandlerRegistry::new(), + ) + .await + .expect("accept b"); + + let event_a = events_a.recv().await.expect("event a"); + let event_b = events_b.recv().await.expect("event b"); + assert_eq!( + event_a, + SessionEvent::HandshakeNegotiated { + version: 1, + shared_domains: vec!["core/exec".to_owned()], + } + ); + assert_eq!(event_a, event_b); + + session_a.close().await.expect("close a"); + session_b.close().await.expect("close b"); + } + + #[tokio::test] + async fn a_manage_request_with_no_registered_handler_gets_an_explicit_unknown_verb_error() { + let (a, b) = tcp_pair().await; + let (session_a, _events_a) = Session::accept(a, [], HandlerRegistry::new()) + .await + .expect("accept a"); + let (_session_b, mut _events_b) = Session::accept(b, [], HandlerRegistry::new()) + .await + .expect("accept b"); + + let outcome = session_a + .send_manage_request( + json_command("example.com/nothing:here"), + scope(), + None, + None, + ) + .await + .expect("send_manage_request"); + match outcome { + ManageOutcome::Error(error) => assert_eq!(error.code, "unknown-verb"), + ManageOutcome::Ok(_) => panic!("expected an unknown-verb error"), + } + } + + #[tokio::test] + async fn a_registered_handler_dispatches_and_the_caller_receives_its_outcome() { + let (a, b) = tcp_pair().await; + let registry = + HandlerRegistry::new().register("example.com/echo:ping", Arc::new(EchoHandler)); + let (session_a, _events_a) = Session::accept(a, [], HandlerRegistry::new()) + .await + .expect("accept a"); + let (_session_b, _events_b) = Session::accept(b, [], registry).await.expect("accept b"); + + let outcome = session_a + .send_manage_request(json_command("example.com/echo:ping"), scope(), None, None) + .await + .expect("send_manage_request"); + match outcome { + ManageOutcome::Ok(ok) => { + assert!(ok.extra.get(&"echoed-request-id".to_owned()).is_some()); + } + ManageOutcome::Error(error) => panic!("expected ok, got {error:?}"), + } + } + + /// A handler that never resolves -- `send_manage_request`'s own + /// timeout is the only thing that can end the wait, proving the + /// timeout is real and not merely returned instantly by coincidence. + struct NeverRespondsHandler; + + #[async_trait::async_trait] + impl ManageRequestHandler for NeverRespondsHandler { + async fn handle(&self, _request: IncomingManageRequest) -> ManageOutcome { + std::future::pending::<()>().await; + unreachable!("pending() never resolves") + } + } + + #[tokio::test] + async fn send_manage_request_times_out_when_no_response_arrives() { + let (a, b) = tcp_pair().await; + let registry = HandlerRegistry::new() + .register("example.com/silence:ping", Arc::new(NeverRespondsHandler)); + let (session_a, _events_a) = Session::accept(a, [], HandlerRegistry::new()) + .await + .expect("accept a"); + let (_session_b, _events_b) = Session::accept(b, [], registry).await.expect("accept b"); + + let outcome = session_a + .send_manage_request( + json_command("example.com/silence:ping"), + scope(), + None, + Some(Duration::from_millis(50)), + ) + .await + .expect("send_manage_request"); + match outcome { + ManageOutcome::Error(error) => assert_eq!(error.code, "timeout"), + ManageOutcome::Ok(_) => panic!("expected a timeout error"), + } + } + + #[tokio::test] + async fn closing_a_session_rejects_a_pending_request_and_emits_disconnected() { + let (a, b) = tcp_pair().await; + let registry = HandlerRegistry::new() + .register("example.com/silence:ping", Arc::new(NeverRespondsHandler)); + let (session_a, mut events_a) = Session::accept(a, [], HandlerRegistry::new()) + .await + .expect("accept a"); + let (session_b, _events_b) = Session::accept(b, [], registry).await.expect("accept b"); + + let pending = tokio::spawn({ + let session_a = std::sync::Arc::new(session_a); + let session_a_for_request = std::sync::Arc::clone(&session_a); + async move { + let outcome = session_a_for_request + .send_manage_request( + json_command("example.com/silence:ping"), + scope(), + None, + None, + ) + .await; + (outcome, session_a) + } + }); + + // Give the request time to actually be in flight before closing. + tokio::time::sleep(Duration::from_millis(50)).await; + session_b.close().await.expect("close b"); + + let (result, session_a) = pending.await.expect("join"); + assert!( + result.is_err(), + "expected the disconnect to reject the pending request" + ); + session_a.close().await.expect("close a"); + + let mut saw_disconnected = false; + while let Ok(event) = events_a.try_recv() { + if matches!(event, SessionEvent::Disconnected { .. }) { + saw_disconnected = true; + } + } + assert!( + saw_disconnected, + "expected a Disconnected event on session a" + ); + } + + #[tokio::test] + async fn send_data_frame_delivers_the_frame_directly() { + let (a, mut b) = tcp_pair().await; + let (session_a, _events_a) = Session::accept(a, [], HandlerRegistry::new()) + .await + .expect("accept a"); + + // b is not wrapped as a Session here -- reading its raw frame stream directly proves send_data_frame puts the frame on the wire unmodified, with no manage-request/response envelope. + let mut raw = b.receive().expect("receive"); + // Consume b's own inbound handshake frame (sent by session_a's accept) before asserting on the data frame. + let _ = next_frame(&mut raw).await; + + let peer = DeviceId::from_bytes([7u8; 32]); + session_a + .send_data_frame(DataFrame::Have(DataHaveFrame { peer, head_seq: 42 })) + .await + .expect("send_data_frame"); + + match next_frame(&mut raw).await { + Some(Frame::DataHave(frame)) => { + assert_eq!(frame.peer, peer); + assert_eq!(frame.head_seq, 42); + } + other => panic!("expected data-have, got {other:?}"), + } + + session_a.close().await.expect("close a"); + } +} From 47eb11b2d1a92adbd21e1e6f24eface0ab42c492 Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Thu, 17 Sep 2026 22:18:07 +0100 Subject: [PATCH 2/2] test(threshold): prove the threshold signing round trip over the real session runtime Adds an integration test that drives wire-mesh#171's NetworkThresholdCoordinator and handle_threshold_* participant handlers over two real TCP connections wrapped as wire-mesh-core's new Session, rather than the fake in-memory sender/participant test doubles src/network.rs's own unit test uses: a coordinator dials each participant, each participant registers a ManageRequestHandler for the threshold verb, and a full commit/sign round trip produces a signature that verifies against the DKG group's public key. Adds wire-mesh-core as a dev-dependency with default (net-on) features specifically for this test binary, alongside the existing net-off [dependencies] entry the wasm32-unknown-unknown build still relies on -- Cargo unifies dev-dependency features only for this crate's own test builds, never for a consumer (wire-mesh-threshold-wasm) that depends on this crate as an ordinary dependency and never builds its tests. --- rust/crates/wire-mesh-threshold/Cargo.toml | 2 + .../tests/session_runtime_network.rs | 371 ++++++++++++++++++ 2 files changed, 373 insertions(+) create mode 100644 rust/crates/wire-mesh-threshold/tests/session_runtime_network.rs diff --git a/rust/crates/wire-mesh-threshold/Cargo.toml b/rust/crates/wire-mesh-threshold/Cargo.toml index 5e70a8a..1cad0c6 100644 --- a/rust/crates/wire-mesh-threshold/Cargo.toml +++ b/rust/crates/wire-mesh-threshold/Cargo.toml @@ -23,6 +23,8 @@ async-trait = { workspace = true } [dev-dependencies] tokio = { workspace = true } +# Test-only, net feature ON: unlike the [dependencies] entry above (net off, for the wasm32-unknown-unknown build), the integration test in tests/session_runtime_network.rs proves the real wiring of this crate's ManageRequestSender/handle_threshold_* onto wire-mesh-core's session runtime, which needs a live tokio task-spawning runtime. Cargo unifies this crate's own feature requests across [dependencies]/[dev-dependencies] for a test build only -- wire-mesh-threshold-wasm, which depends on this crate as an ordinary (non-dev) dependency and never builds this crate's own tests, never sees "net" turned on. +wire-mesh-core = { path = "../wire-mesh-core" } [lints] workspace = true diff --git a/rust/crates/wire-mesh-threshold/tests/session_runtime_network.rs b/rust/crates/wire-mesh-threshold/tests/session_runtime_network.rs new file mode 100644 index 0000000..85e1614 --- /dev/null +++ b/rust/crates/wire-mesh-threshold/tests/session_runtime_network.rs @@ -0,0 +1,371 @@ +//! Real, end-to-end proof that wire-mesh#173's session runtime is +//! actually usable by wire-mesh#171's threshold network integration -- +//! not against `FakeSender`/`FakeParticipant` test doubles (as +//! `src/network.rs`'s own unit test already covers), but over two real +//! TCP connections carrying real `manage-request`/`manage-response` +//! frames through `wire_mesh_core::domain::session::Session`. +//! +//! `wire-mesh-core` is a normal ([dependencies]) dependency of this crate +//! with its "net" feature off, so `wire-mesh-threshold-wasm`'s +//! wasm32-unknown-unknown build never pulls in tokio's networking stack +//! (see this crate's own Cargo.toml comment). This test file is built +//! only for `cargo test`, which unifies this crate's own +//! `[dev-dependencies]` (declaring `wire-mesh-core` with its default, +//! "net"-on, features) into the test binary's dependency graph -- a +//! unification that never reaches the wasm crate, since it depends on +//! `wire-mesh-threshold` as an ordinary dependency and never builds this +//! crate's own tests. + +use std::collections::HashMap; +use std::sync::{Arc, Mutex as StdMutex}; +use std::time::Duration; + +use frost_ed25519::keys::{KeyPackage, PublicKeyPackage}; +use wire_mesh_core::adapters::node_identity::{derive_device_id_from_public_key, NodeIdentity}; +use wire_mesh_core::domain::session::{ + HandlerRegistry, IncomingManageRequest, ManageRequestHandler, Session, +}; +use wire_mesh_core::ports::{Connection, CoreError, Identity, OnConnection, Transport}; +use wire_mesh_threshold::dkg::{round1 as dkg_round1, round2 as dkg_round2, round3 as dkg_round3}; +use wire_mesh_threshold::identifiers::identifier_for_device; +use wire_mesh_threshold::identity::ThresholdCoordinator; +use wire_mesh_threshold::network::{ + handle_threshold_abort, handle_threshold_commit, handle_threshold_sign, ManageRequestSender, + NetworkThresholdCoordinator, SigningSessionState, THRESHOLD_SIGN_VERB, +}; +use wire_mesh_threshold::nonce_store::InMemoryNonceStore; +use wire_mesh_threshold::signing::{aggregate, build_signing_package}; +use wire_mesh_threshold::subject::{to_be_signed, ThresholdSubject}; +use wire_mesh_wire::identity::DeviceId; +use wire_mesh_wire::management::{ + ManageCommand, ManageError, ManageOk, ManageOutcome, ManageParams, +}; +use wire_mesh_wire::tokens::{CapabilityScope, CoseSign1}; +use wire_mesh_wire::value::{CanonicalMap, CborValue}; + +fn now_unix_ms() -> u64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .expect("system clock is after the Unix epoch") + .as_millis() as u64 +} + +fn manage_ok(fields: Vec<(&str, Vec)>) -> ManageOutcome { + let mut extra = CanonicalMap::new(); + for (key, value) in fields { + extra + .insert(key.to_owned(), CborValue::Bytes(value)) + .expect("unique key"); + } + ManageOutcome::Ok(ManageOk { extra }) +} + +fn manage_error(code: &str) -> ManageOutcome { + ManageOutcome::Error(ManageError { + code: code.to_owned(), + message: None, + }) +} + +/// A fresh T=2-of-2 DKG purely as test fixture setup, identical in shape +/// to `src/network.rs`'s own `dkg_fixture` (private to that module, so +/// reproduced here rather than reused) -- see that copy's own doc comment +/// for why `device_ids` must be each participant's REAL personal +/// device-id. +fn dkg_fixture( + device_ids: &[DeviceId], +) -> ( + HashMap, + DeviceId, + PublicKeyPackage, +) { + use std::collections::BTreeMap as Map; + + let ids: Vec = device_ids + .iter() + .map(|d| identifier_for_device(d).expect("derives")) + .collect(); + + let mut secrets1 = Map::new(); + let mut packages1: Map = + Map::new(); + for &id in &ids { + let (s, p) = dkg_round1(id, ids.len() as u16, 2).expect("round1"); + secrets1.insert(id, s); + packages1.insert(id, p); + } + + let mut round2_inbox: Map< + frost_ed25519::Identifier, + Map, + > = Map::new(); + let mut secrets2 = Map::new(); + for &id in &ids { + let own = secrets1.remove(&id).expect("secret1"); + let others: Map<_, _> = packages1 + .iter() + .filter(|(&p, _)| p != id) + .map(|(&p, v)| (p, v.clone())) + .collect(); + let (s2, outgoing) = dkg_round2(own, &others).expect("round2"); + secrets2.insert(id, s2); + for (recipient, package) in outgoing { + round2_inbox + .entry(recipient) + .or_default() + .insert(id, package); + } + } + + let mut key_packages = HashMap::new(); + let mut group_device_id = None; + let mut group_public_key_package = None; + for &id in &ids { + let s2 = secrets2.get(&id).expect("secret2"); + let others1: Map<_, _> = packages1 + .iter() + .filter(|(&p, _)| p != id) + .map(|(&p, v)| (p, v.clone())) + .collect(); + let inbox = round2_inbox.get(&id).expect("inbox"); + let (kp, pkp) = dkg_round3(s2, &others1, inbox).expect("round3"); + group_device_id = Some(derive_device_id_from_public_key( + &pkp.verifying_key().serialize().expect("serialize"), + )); + group_public_key_package = Some(pkp); + key_packages.insert(id, kp); + } + + ( + key_packages, + group_device_id.expect("at least one participant"), + group_public_key_package.expect("at least one participant"), + ) +} + +/// The participant side of the wire-mesh#171 protocol, wired onto +/// wire-mesh#173's session runtime as a +/// [`wire_mesh_core::domain::session::ManageRequestHandler`] registered +/// for [`THRESHOLD_SIGN_VERB`]: dispatches an incoming manage-request's +/// decoded `ManageParams` variant to the matching `handle_threshold_*` +/// function, exactly the wiring `network.rs`'s own module doc comment +/// says this codebase lacked before this session runtime existed. +struct ThresholdParticipant { + own_device_id: DeviceId, + key_package: KeyPackage, + identity: NodeIdentity, + group: DeviceId, + nonce_store: InMemoryNonceStore, + session_state: StdMutex>, +} + +#[async_trait::async_trait] +impl ManageRequestHandler for ThresholdParticipant { + async fn handle(&self, request: IncomingManageRequest) -> ManageOutcome { + match request.command.params { + ManageParams::ThresholdCommit(params) => { + let result = handle_threshold_commit( + ¶ms, + self.own_device_id, + |group| { + if *group == self.group { + Some(self.key_package.clone()) + } else { + None + } + }, + &self.nonce_store, + now_unix_ms(), + |_subject: &ThresholdSubject| None, + ); + match result { + Ok((response, state)) => { + self.session_state + .lock() + .unwrap_or_else(|e| e.into_inner()) + .insert(params.session_id, state); + manage_ok(vec![ + ("participant", response.participant.as_ref().to_vec()), + ("hiding", response.hiding), + ("binding", response.binding), + ]) + } + Err(error) => manage_error(&error.to_string()), + } + } + ManageParams::ThresholdSign(params) => { + let state = self + .session_state + .lock() + .unwrap_or_else(|e| e.into_inner()) + .remove(¶ms.session_id); + let Some(state) = state else { + return manage_error("no-such-session"); + }; + let result = handle_threshold_sign( + ¶ms, + &state, + &self.key_package, + &self.nonce_store, + &self.identity, + self.group, + ) + .await; + match result { + Ok(envelope) => manage_ok(vec![("share", envelope.encode_to_vec())]), + Err(error) => manage_error(&error.to_string()), + } + } + ManageParams::ThresholdAbort(params) => { + handle_threshold_abort(params.session_id, &self.nonce_store); + manage_ok(vec![]) + } + _ => manage_error("unsupported"), + } + } +} + +/// The coordinator side: a [`ManageRequestSender`] that routes each +/// outbound manage-request to the [`Session`] already dialled to that +/// target device, proving [`NetworkThresholdCoordinator`] (built and +/// tested in `src/network.rs` against a fake sender) drives a real +/// session exactly the same way. +struct SessionManageRequestSender { + sessions: HashMap, +} + +#[async_trait::async_trait] +impl ManageRequestSender for SessionManageRequestSender { + async fn send_manage_request( + &self, + target: DeviceId, + command: ManageCommand, + scope: CapabilityScope, + token: Option, + ) -> Result { + let session = self + .sessions + .get(&target) + .expect("test fixture: unknown target device"); + session + .send_manage_request(command, scope, token, Some(Duration::from_secs(5))) + .await + } +} + +#[tokio::test] +async fn threshold_signing_round_trips_over_real_tcp_sessions() { + // Personal identities are generated FIRST, and the DKG fixture is + // built around their own real device-ids -- see the fixture's own + // doc comment for why a synthetic placeholder device-id would + // silently derive the wrong FROST identifier. + let personal_identities: Vec = + (0..2).map(|_| NodeIdentity::generate_ed25519()).collect(); + let device_ids: Vec = personal_identities + .iter() + .map(|identity| *identity.device_id()) + .collect(); + let (key_packages, group, public_key_package) = dkg_fixture(&device_ids); + + let transport = wire_mesh_core::adapters::TcpTransport::new(); + + // One TCP listener per participant, kept alive for the whole test + // via `_listen_guards` (dropping a guard stops that listener). + let mut listen_guards = Vec::new(); + let mut participant_addrs = Vec::new(); + let mut accepted_rxs = Vec::new(); + for _ in &device_ids { + let (tx, rx) = tokio::sync::mpsc::unbounded_channel::>(); + let on_connection: OnConnection = Arc::new(move |connection| { + let _ = tx.send(connection); + }); + let guard = transport + .listen("127.0.0.1:0", on_connection) + .await + .expect("listen"); + participant_addrs.push(guard.address()); + listen_guards.push(guard); + accepted_rxs.push(rx); + } + + // The coordinator dials every participant and wires each connection + // up as a Session -- it registers no handlers of its own, since it + // never receives a manage-request in this protocol, only sends them. + let mut coordinator_sessions = HashMap::new(); + for (device_id, addr) in device_ids.iter().zip(participant_addrs.iter()) { + let connection = transport.connect(addr).await.expect("connect"); + let (session, _events) = Session::accept(connection, [], HandlerRegistry::new()) + .await + .expect("coordinator-side accept"); + coordinator_sessions.insert(*device_id, session); + } + + // Each participant accepts its inbound connection from the + // coordinator and wires it up with a real ThresholdParticipant + // handler registered for THRESHOLD_SIGN_VERB. The returned Session/ + // event-receiver pair is intentionally dropped: the dispatch loop + // that actually answers requests runs as an independent, already- + // spawned background task holding its own connection/registry + // handles, so nothing here needs to keep the caller's own Session + // value alive for the loop to keep running. + for ((device_id, identity), mut accepted_rx) in device_ids + .iter() + .zip(personal_identities.into_iter()) + .zip(accepted_rxs.into_iter()) + { + let connection = accepted_rx.recv().await.expect("accepted connection"); + let identifier = identifier_for_device(device_id).expect("derives"); + let key_package = key_packages.get(&identifier).expect("key package").clone(); + let handler = Arc::new(ThresholdParticipant { + own_device_id: *device_id, + key_package, + identity, + group, + nonce_store: InMemoryNonceStore::new(), + session_state: StdMutex::new(HashMap::new()), + }); + let registry = HandlerRegistry::new().register(THRESHOLD_SIGN_VERB, handler); + let (_session, _events) = Session::accept(connection, [], registry) + .await + .expect("participant-side accept"); + } + + let sender = SessionManageRequestSender { + sessions: coordinator_sessions, + }; + let verifier_identity: Arc = Arc::new(NodeIdentity::generate_ed25519()); + let coordinator = NetworkThresholdCoordinator::new(sender, group, verifier_identity); + + let subject = ThresholdSubject { + kind: "capability-token".to_owned(), + protected: vec![0xa1, 0x01, 0x27], + payload: vec![0xa1, 0x00, 0x01], + }; + let message = to_be_signed(&subject); + + let commitments = coordinator + .commit_round(1, &device_ids, &subject, u64::MAX) + .await + .expect("commit_round"); + assert_eq!(commitments.len(), 2, "both participants committed"); + + let shares = coordinator + .sign_round(1, &commitments) + .await + .expect("sign_round"); + assert_eq!(shares.len(), 2, "both participants released a share"); + + let commitments_map: std::collections::BTreeMap<_, _> = commitments.into_iter().collect(); + let signing_package = build_signing_package(commitments_map, &message); + let shares_map: std::collections::BTreeMap<_, _> = shares.into_iter().collect(); + + let signature = + aggregate(&signing_package, &shares_map, &public_key_package).expect("aggregate"); + assert!( + public_key_package + .verifying_key() + .verify(&message, &signature) + .is_ok(), + "the aggregated signature verifies against the group's public key" + ); +}