diff --git a/common/client-core/src/lib.rs b/common/client-core/src/lib.rs index 1f2faf3659..e9f0494039 100644 --- a/common/client-core/src/lib.rs +++ b/common/client-core/src/lib.rs @@ -1,3 +1,4 @@ +#![allow(deprecated)] // silences clippy warning: use of deprecated associated function `nym_crypto::generic_array::GenericArray::::clone_from_slice`: please upgrade to generic-array 1.x - TODO use std::future::Future; #[cfg(all( diff --git a/common/client-core/surb-storage/src/lib.rs b/common/client-core/surb-storage/src/lib.rs index 079c213bbc..ffd7ad85f6 100644 --- a/common/client-core/surb-storage/src/lib.rs +++ b/common/client-core/surb-storage/src/lib.rs @@ -1,6 +1,7 @@ // Copyright 2022 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 +#![allow(deprecated)] // silences clippy warning: use of deprecated associated function `nym_crypto::generic_array::GenericArray::::from_exact_iter`: please upgrade to generic-array 1.x - TODO pub use backend::*; pub use combined::CombinedReplyStorage; pub use key_storage::SentReplyKeys; diff --git a/common/cosmwasm-smart-contracts/coconut-dkg/src/types.rs b/common/cosmwasm-smart-contracts/coconut-dkg/src/types.rs index 6e4fdef06a..20d4920db6 100644 --- a/common/cosmwasm-smart-contracts/coconut-dkg/src/types.rs +++ b/common/cosmwasm-smart-contracts/coconut-dkg/src/types.rs @@ -1,6 +1,8 @@ // Copyright 2022-2023 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 +#![allow(clippy::derivable_impls)] +// MAX: surpressing warning for the moment, will be dealt with in a different PR (TODO) use cosmwasm_schema::cw_serde; use std::fmt::{Display, Formatter}; use std::str::FromStr; diff --git a/common/crypto/src/lib.rs b/common/crypto/src/lib.rs index 1dff7b82be..e8426bb085 100644 --- a/common/crypto/src/lib.rs +++ b/common/crypto/src/lib.rs @@ -1,6 +1,8 @@ // Copyright 2021 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 +#![allow(deprecated)] // silences clippy warning: deprecated associated function `generic_array::GenericArray::::from_exact_iter`: please upgrade to generic-array 1.x - TODO + #[cfg(feature = "asymmetric")] pub mod asymmetric; pub mod bech32_address_validation; diff --git a/common/gateway-requests/src/lib.rs b/common/gateway-requests/src/lib.rs index b63b29dca1..e577cb0e5a 100644 --- a/common/gateway-requests/src/lib.rs +++ b/common/gateway-requests/src/lib.rs @@ -1,5 +1,6 @@ // Copyright 2020-2022 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 +#![allow(deprecated)] // silences clippy warning: deprecated associated function `nym_crypto::generic_array::GenericArray::::clone_from_slice`: please upgrade to generic-array 1.x - TODO pub use nym_crypto::generic_array; use nym_crypto::OutputSizeUser; diff --git a/common/http-api-client/src/lib.rs b/common/http-api-client/src/lib.rs index d39e77e6e5..c9baec7eb7 100644 --- a/common/http-api-client/src/lib.rs +++ b/common/http-api-client/src/lib.rs @@ -1,6 +1,9 @@ // Copyright 2023 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 +#![allow(deprecated)] +// silences clippy warning: use of deprecated tuple variant `HttpClientError::GenericRequestFailure`: use another more strongly typed variant - this variant is only left for compatibility reasons - TODO + //! Nym HTTP API Client //! //! Centralizes and implements the core API client functionality. This crate provides custom, diff --git a/common/nymsphinx/acknowledgements/src/lib.rs b/common/nymsphinx/acknowledgements/src/lib.rs index 490e94eaea..d37990d49f 100644 --- a/common/nymsphinx/acknowledgements/src/lib.rs +++ b/common/nymsphinx/acknowledgements/src/lib.rs @@ -1,5 +1,6 @@ // Copyright 2021 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 +#![allow(deprecated)] // silences clippy warning: deprecated associated function `nym_crypto::generic_array::GenericArray::::clone_from_slice`: please upgrade to generic-array 1.x - TODO pub mod identifier; pub mod key; diff --git a/common/nymsphinx/anonymous-replies/src/lib.rs b/common/nymsphinx/anonymous-replies/src/lib.rs index 2c47a23a76..84bfa7a323 100644 --- a/common/nymsphinx/anonymous-replies/src/lib.rs +++ b/common/nymsphinx/anonymous-replies/src/lib.rs @@ -1,6 +1,7 @@ // Copyright 2021 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 +#![allow(deprecated)] // silences clippy warning: deprecated struct `nym_crypto::generic_array::GenericArray`: please upgrade to generic-array 1.x - TODO pub mod encryption_key; pub mod reply_surb; pub mod requests; diff --git a/common/store-cipher/src/lib.rs b/common/store-cipher/src/lib.rs index 98d586151f..5119021d16 100644 --- a/common/store-cipher/src/lib.rs +++ b/common/store-cipher/src/lib.rs @@ -1,6 +1,7 @@ // Copyright 2023 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 +#![allow(deprecated)] use aes_gcm::aead::{Aead, Nonce}; use aes_gcm::{AeadCore, AeadInPlace, KeyInit}; use rand::{thread_rng, CryptoRng, Fill, RngCore}; diff --git a/documentation/docs/pnpm-lock.yaml b/documentation/docs/pnpm-lock.yaml index f880137e0c..bac8e0871d 100644 --- a/documentation/docs/pnpm-lock.yaml +++ b/documentation/docs/pnpm-lock.yaml @@ -104,6 +104,9 @@ importers: next: specifier: 15.5.10 version: 15.5.10(@opentelemetry/api@1.9.0)(babel-plugin-macros@3.1.0)(react-dom@18.3.1(react@18.3.1))(react@18.3.1) + next-sitemap: + specifier: 4.2.3 + version: 4.2.3(next@15.5.10(@opentelemetry/api@1.9.0)(babel-plugin-macros@3.1.0)(react-dom@18.3.1(react@18.3.1))(react@18.3.1)) nextra: specifier: '2' version: 2.13.4(next@15.5.10(@opentelemetry/api@1.9.0)(babel-plugin-macros@3.1.0)(react-dom@18.3.1(react@18.3.1))(react@18.3.1))(react-dom@18.3.1(react@18.3.1))(react@18.3.1) diff --git a/nym-outfox/src/lib.rs b/nym-outfox/src/lib.rs index e10d30e846..0d278c06d7 100644 --- a/nym-outfox/src/lib.rs +++ b/nym-outfox/src/lib.rs @@ -1,3 +1,6 @@ +// MAX: temp ignore deprecated, can be dealt with in its own PR +#![allow(deprecated)] // silences clippy warning: deprecated associated function `chacha20::cipher::generic_array::GenericArray::::from_slice`: please upgrade to generic-array 1.x - TODO + pub mod constants; pub mod error; pub mod format; diff --git a/sdk/rust/nym-sdk/Cargo.toml b/sdk/rust/nym-sdk/Cargo.toml index 70e1441e8a..59bd347d2f 100644 --- a/sdk/rust/nym-sdk/Cargo.toml +++ b/sdk/rust/nym-sdk/Cargo.toml @@ -83,7 +83,7 @@ anyhow = { workspace = true } dotenvy = { workspace = true } reqwest = { workspace = true, features = ["json", "socks"] } thiserror = { workspace = true } -tokio = { workspace = true, features = ["full"] } +tokio = { workspace = true, features = ["full", "test-util"] } time = { workspace = true } nym-bin-common = {workspace = true, features = ["basic_tracing"] } diff --git a/sdk/rust/nym-sdk/README.md b/sdk/rust/nym-sdk/README.md index 16f1989414..be8bdab8ed 100644 --- a/sdk/rust/nym-sdk/README.md +++ b/sdk/rust/nym-sdk/README.md @@ -1,5 +1,7 @@ # Nym Rust SDK + + This repo contains several components: - `mixnet`: exposes Nym Client builders and methods. This is useful if you want to interact directly with the Client, or build transport abstractions. - `tcp_proxy`: exposes functionality to set up client/server instances that expose a localhost TcpSocket to read/write to like a 'normal' socket connection. `tcp_proxy/bin/` contains standalone `nym-proxy-client` and `nym-proxy-server` binaries. diff --git a/sdk/rust/nym-sdk/examples/stream_idle_timeout.rs b/sdk/rust/nym-sdk/examples/stream_idle_timeout.rs new file mode 100644 index 0000000000..0e4b84565b --- /dev/null +++ b/sdk/rust/nym-sdk/examples/stream_idle_timeout.rs @@ -0,0 +1,78 @@ +// Copyright 2024 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +//! Demonstrates stream idle timeout cleanup. +//! +//! Opens a stream to self, uses it, then stops. After the idle timeout +//! elapses the router removes the stream and reads return EOF. +//! +//! Run with: cargo run --example stream_idle_timeout + +use nym_sdk::mixnet; +use std::time::Duration; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; + +/// Short idle timeout so we don't wait the default +const IDLE_TIMEOUT: Duration = Duration::from_secs(2); +const WAIT_TIMEOUT: Duration = Duration::from_secs(60); + +#[tokio::main] +async fn main() { + nym_bin_common::logging::setup_tracing_logger(); + + // Build a client with a short stream idle timeout. + let mut client = mixnet::MixnetClientBuilder::new_ephemeral() + .with_stream_idle_timeout(IDLE_TIMEOUT) + .build() + .unwrap() + .connect_to_mixnet() + .await + .unwrap(); + + let our_address = *client.nym_address(); + println!("Client address: {our_address}"); + + // Set up a listener and open a stream to ourselves. + let mut listener = client.listener().unwrap(); + let mut outbound = client.open_stream(our_address, None).await.unwrap(); + println!("Opened outbound stream: {}", outbound.id()); + + let mut inbound = tokio::time::timeout(WAIT_TIMEOUT, listener.accept()) + .await + .expect("timed out waiting for accept") + .expect("listener shut down"); + println!("Accepted inbound stream: {}", inbound.id()); + + // Use the stream. + let msg = b"hello from idle timeout example"; + outbound.write_all(msg).await.unwrap(); + outbound.flush().await.unwrap(); + + let mut buf = vec![0u8; 1024]; + let n = tokio::time::timeout(WAIT_TIMEOUT, inbound.read(&mut buf)) + .await + .expect("timed out reading") + .expect("read failed"); + println!("Received: {:?}", String::from_utf8_lossy(&buf[..n])); + assert_eq!(&buf[..n], msg); + + // Now stop using the stream and wait for the idle timeout. + println!( + "\nStream is idle. Waiting {}s for cleanup...", + IDLE_TIMEOUT.as_secs() + ); + tokio::time::sleep(IDLE_TIMEOUT + Duration::from_secs(2)).await; + + // The router should have cleaned up the stream. The inbound receiver + // is closed, so a read returns 0 bytes (EOF). + let n = inbound.read(&mut buf).await.expect("read failed"); + if n == 0 { + println!("Inbound stream returned EOF — cleaned up by idle timeout."); + } else { + println!("Unexpected: got {n} bytes after idle timeout"); + } + + drop(outbound); + drop(inbound); + client.disconnect().await; +} diff --git a/sdk/rust/nym-sdk/examples/stream_mode_guard.rs b/sdk/rust/nym-sdk/examples/stream_mode_guard.rs new file mode 100644 index 0000000000..c57d7e3a3d --- /dev/null +++ b/sdk/rust/nym-sdk/examples/stream_mode_guard.rs @@ -0,0 +1,69 @@ +// Copyright 2024 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use nym_sdk::mixnet; +use nym_sdk::mixnet::MixnetMessageSender; +use nym_sdk::Error; + +#[tokio::main] +async fn main() { + nym_bin_common::logging::setup_tracing_logger(); + + let mut client = mixnet::MixnetClient::connect_new().await.unwrap(); + let our_address = *client.nym_address(); + println!("Our client nym address is: {our_address}"); + + // Message-based API works before stream mode is activated + println!("\nTesting message-based API (should work)"); + client + .send_plain_message(our_address, "hello via message API") + .await + .unwrap(); + println!("Message sent successfully via message-based API"); + + // Now activate stream mode by creating a listener + println!("\nActivating stream mode via listener()"); + let _listener = client.listener().unwrap(); + println!("Stream mode is now active"); + + // Message-based API should now fail + println!("\nTesting message-based API again (should fail)"); + let result = client + .send_plain_message(our_address, "this should fail") + .await; + + match result { + Err(Error::StreamModeActive) => { + println!("Got expected error: StreamModeActive"); + println!("Message-based API correctly blocked after stream mode activation"); + } + Err(e) => { + println!("Got unexpected error: {e:?}"); + } + Ok(()) => { + println!("ERROR: send() should have failed but succeeded!"); + } + } + + // split_sender shares the stream_mode flag + println!("\nTesting split_sender (shares stream_mode)"); + let sender = client.split_sender(); + let result = sender + .send_plain_message(our_address, "this should also fail") + .await; + + match result { + Err(Error::StreamModeActive) => { + println!("Got expected error: StreamModeActive on split sender"); + println!("Split sender correctly shares stream_mode with parent client"); + } + Err(e) => { + println!("Got unexpected error: {e:?}"); + } + Ok(()) => { + println!("ERROR: split_sender.send() should have failed but succeeded!"); + } + } + + client.disconnect().await; +} diff --git a/sdk/rust/nym-sdk/examples/stream_simple_read_write.rs b/sdk/rust/nym-sdk/examples/stream_simple_read_write.rs new file mode 100644 index 0000000000..480297f5ad --- /dev/null +++ b/sdk/rust/nym-sdk/examples/stream_simple_read_write.rs @@ -0,0 +1,151 @@ +// Copyright 2024 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +//! Demonstrates concurrent streams over the Mixnet. +//! +//! One sender opens streams to two receivers. +//! Both receivers accept, read, and reply concurrently. +//! +//! Run with: cargo run --example async_read_write + +use nym_sdk::mixnet; +use std::time::Duration; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; + +const TIMEOUT: Duration = Duration::from_secs(60); + +#[tokio::main] +async fn main() { + nym_bin_common::logging::setup_tracing_logger(); + + let mut sender = mixnet::MixnetClient::connect_new().await.unwrap(); + println!("Sender address: {}", sender.nym_address()); + + let mut receiver_a = mixnet::MixnetClient::connect_new().await.unwrap(); + let addr_a = *receiver_a.nym_address(); + println!("Receiver A address: {addr_a}"); + + let mut receiver_b = mixnet::MixnetClient::connect_new().await.unwrap(); + let addr_b = *receiver_b.nym_address(); + println!("Receiver B address: {addr_b}"); + + let mut listener_a = receiver_a.listener().unwrap(); + let mut listener_b = receiver_b.listener().unwrap(); + + println!("\nOpening streams to both receivers..."); + let mut stream_to_a = sender.open_stream(addr_a, None).await.unwrap(); + println!("Stream to A opened: {}", stream_to_a.id()); + + let mut stream_to_b = sender.open_stream(addr_b, None).await.unwrap(); + println!("Stream to B opened: {}", stream_to_b.id()); + + println!("\nWaiting for both receivers to accept..."); + let (inbound_a, inbound_b) = tokio::try_join!( + async { + tokio::time::timeout(TIMEOUT, listener_a.accept()) + .await + .expect("timed out waiting for A to accept") + .ok_or("listener A shut down") + }, + async { + tokio::time::timeout(TIMEOUT, listener_b.accept()) + .await + .expect("timed out waiting for B to accept") + .ok_or("listener B shut down") + }, + ) + .unwrap(); + println!("A accepted stream: {}", inbound_a.id()); + println!("B accepted stream: {}", inbound_b.id()); + + let msg_a = b"hello receiver A"; + let msg_b = b"hello receiver B"; + + println!("\nSender writing to both streams..."); + stream_to_a.write_all(msg_a).await.unwrap(); + stream_to_a.flush().await.unwrap(); + + stream_to_b.write_all(msg_b).await.unwrap(); + stream_to_b.flush().await.unwrap(); + + println!("\nBoth receivers reading and replying concurrently..."); + let reply_a = b"reply from A"; + let reply_b = b"reply from B"; + + let (res_a, res_b) = tokio::join!( + // Receiver A: read then reply + async { + let mut inbound = inbound_a; + let mut buf = vec![0u8; 1024]; + let n = tokio::time::timeout(TIMEOUT, inbound.read(&mut buf)) + .await + .expect("A: timed out reading") + .expect("A: read failed"); + println!("Receiver A got: {:?}", String::from_utf8_lossy(&buf[..n])); + assert_eq!(&buf[..n], msg_a); + + inbound.write_all(reply_a).await.unwrap(); + inbound.flush().await.unwrap(); + println!("Receiver A replied"); + inbound + }, + // Receiver B: read then reply + async { + let mut inbound = inbound_b; + let mut buf = vec![0u8; 1024]; + let n = tokio::time::timeout(TIMEOUT, inbound.read(&mut buf)) + .await + .expect("B: timed out reading") + .expect("B: read failed"); + println!("Receiver B got: {:?}", String::from_utf8_lossy(&buf[..n])); + assert_eq!(&buf[..n], msg_b); + + inbound.write_all(reply_b).await.unwrap(); + inbound.flush().await.unwrap(); + println!("Receiver B replied"); + inbound + }, + ); + let inbound_a = res_a; + let inbound_b = res_b; + + println!("\nSender reading replies..."); + tokio::join!( + async { + let mut buf = vec![0u8; 1024]; + let n = tokio::time::timeout(TIMEOUT, stream_to_a.read(&mut buf)) + .await + .expect("timed out reading reply from A") + .expect("read failed"); + println!( + "Sender got from A: {:?}", + String::from_utf8_lossy(&buf[..n]) + ); + assert_eq!(&buf[..n], reply_a); + }, + async { + let mut buf = vec![0u8; 1024]; + let n = tokio::time::timeout(TIMEOUT, stream_to_b.read(&mut buf)) + .await + .expect("timed out reading reply from B") + .expect("read failed"); + println!( + "Sender got from B: {:?}", + String::from_utf8_lossy(&buf[..n]) + ); + assert_eq!(&buf[..n], reply_b); + }, + ); + + println!("\nConcurrent round-trips successful!"); + + // Streams clean up on drop (unregister from router). + // No close message is sent over the wire — see stream.rs. + drop(stream_to_a); + drop(stream_to_b); + drop(inbound_a); + drop(inbound_b); + sender.disconnect().await; + receiver_a.disconnect().await; + receiver_b.disconnect().await; +} diff --git a/sdk/rust/nym-sdk/examples/stream_throughput.rs b/sdk/rust/nym-sdk/examples/stream_throughput.rs new file mode 100644 index 0000000000..7b0c9192f3 --- /dev/null +++ b/sdk/rust/nym-sdk/examples/stream_throughput.rs @@ -0,0 +1,81 @@ +// Copyright 2024 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +//! Sends a 1 MB random file over a MixnetStream and verifies the +//! receiver got an identical copy. Cancel with Ctrl+C. +//! +//! Run with: cargo run --example stream_throughput + +use nym_sdk::mixnet; +use rand::RngCore; +use std::time::Duration; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; + +const SIZE: usize = 1024 * 1024; // 1 MB +const TIMEOUT: Duration = Duration::from_secs(300); + +#[tokio::main] +async fn main() { + nym_bin_common::logging::setup_tracing_logger(); + + // Generate random payload + let mut payload = vec![0u8; SIZE]; + rand::rngs::OsRng.fill_bytes(&mut payload); + println!("Generated {} bytes of random data", payload.len()); + + // Connect two clients + println!("Connecting sender..."); + let mut sender = mixnet::MixnetClient::connect_new().await.unwrap(); + println!("{}", sender.nym_address()); + + println!("Connecting receiver..."); + let mut receiver = mixnet::MixnetClient::connect_new().await.unwrap(); + let recv_addr = *receiver.nym_address(); + println!("{recv_addr}"); + + // Open stream + let mut listener = receiver.listener().unwrap(); + let mut tx = sender.open_stream(recv_addr, None).await.unwrap(); + let mut rx = tokio::time::timeout(TIMEOUT, listener.accept()) + .await + .expect("accept timed out") + .expect("listener closed"); + println!("Stream established\n"); + + // Send + let data = payload.clone(); + let send_task = tokio::spawn(async move { + tx.write_all(&data).await.unwrap(); + tx.flush().await.unwrap(); + println!("Sent {} bytes", data.len()); + }); + + // Receive — read exactly SIZE bytes (don't rely on close/EOF - if we need this in future + // iterations we can introduce something like what the TcpProxy module has) + let recv_task = tokio::spawn(async move { + let mut buf = vec![0u8; SIZE]; + tokio::time::timeout(TIMEOUT, rx.read_exact(&mut buf)) + .await + .expect("receive timed out") + .unwrap(); + println!("Received {} bytes", buf.len()); + buf + }); + + let (_, received) = tokio::join!(send_task, recv_task); + let received = received.unwrap(); + + if received == payload { + println!("\nIntegrity OK"); + } else { + eprintln!( + "\nMISMATCH — sent {} bytes, got {}", + payload.len(), + received.len() + ); + std::process::exit(1); + } + + sender.disconnect().await; + receiver.disconnect().await; +} diff --git a/sdk/rust/nym-sdk/src/bandwidth/client.rs b/sdk/rust/nym-sdk/src/bandwidth/client.rs index e1ca588185..fbe2eafbe6 100644 --- a/sdk/rust/nym-sdk/src/bandwidth/client.rs +++ b/sdk/rust/nym-sdk/src/bandwidth/client.rs @@ -29,6 +29,7 @@ where St: Storage + Clone, ::StorageError: Send + Sync + 'static, { + #[allow(clippy::result_large_err)] pub(crate) fn new( network_details: NymNetworkDetails, mnemonic: String, diff --git a/sdk/rust/nym-sdk/src/error.rs b/sdk/rust/nym-sdk/src/error.rs index af7706ec92..57e486172a 100644 --- a/sdk/rust/nym-sdk/src/error.rs +++ b/sdk/rust/nym-sdk/src/error.rs @@ -99,6 +99,15 @@ pub enum Error { #[error("Failed to get shutdown tracker from the task runtime registry: {0}")] RegistryAccess(#[from] nym_task::RegistryAccessError), + + #[error("Cannot use message-based functions after stream mode is activated")] + StreamModeActive, + + #[error("Stream listener has already been taken — listener() can only be called once")] + ListenerAlreadyTaken, + + #[error("Stream subsystem failed to initialise: reconstructed_receiver unavailable")] + StreamInitFailure, } impl Error { diff --git a/sdk/rust/nym-sdk/src/mixnet.rs b/sdk/rust/nym-sdk/src/mixnet.rs index 0ade8cc9e0..ad71e00e3c 100644 --- a/sdk/rust/nym-sdk/src/mixnet.rs +++ b/sdk/rust/nym-sdk/src/mixnet.rs @@ -37,6 +37,7 @@ mod native_client; mod paths; mod sink; mod socks5_client; +pub mod stream; mod traits; pub use client::{DisconnectedMixnetClient, IncludedSurbs, MixnetClientBuilder}; @@ -88,4 +89,5 @@ pub use nym_topology::{provider_trait::TopologyProvider, NymTopology}; pub use paths::StoragePaths; pub use sink::{MixnetMessageSink, MixnetMessageSinkTranslator}; pub use socks5_client::Socks5MixnetClient; +pub use stream::{MixnetListener, MixnetStream, StreamId}; pub use traits::MixnetMessageSender; diff --git a/sdk/rust/nym-sdk/src/mixnet/client.rs b/sdk/rust/nym-sdk/src/mixnet/client.rs index 26f5358568..bafc45b554 100644 --- a/sdk/rust/nym-sdk/src/mixnet/client.rs +++ b/sdk/rust/nym-sdk/src/mixnet/client.rs @@ -42,8 +42,8 @@ use std::sync::Arc; use url::Url; use zeroize::Zeroizing; -// The number of surbs to include in a message by default -const DEFAULT_NUMBER_OF_SURBS: u32 = 10; +/// The number of reply SURBs to include in a message by default. +pub(crate) const DEFAULT_NUMBER_OF_SURBS: u32 = 10; #[derive(Default)] pub struct MixnetClientBuilder { @@ -70,6 +70,7 @@ pub struct MixnetClientBuilder { forget_me: ForgetMe, remember_me: RememberMe, derivation_material: Option, + stream_idle_timeout: Option, } impl MixnetClientBuilder { @@ -112,6 +113,7 @@ impl MixnetClientBuilder { forget_me: Default::default(), remember_me: Default::default(), derivation_material: None, + stream_idle_timeout: None, }) } } @@ -149,6 +151,7 @@ where forget_me: Default::default(), remember_me: Default::default(), derivation_material: None, + stream_idle_timeout: None, } } @@ -175,6 +178,7 @@ where forget_me: self.forget_me, remember_me: self.remember_me, derivation_material: self.derivation_material, + stream_idle_timeout: self.stream_idle_timeout, } } @@ -205,6 +209,15 @@ where self } + /// Set the idle timeout for streams. Streams with no activity for this + /// duration are automatically cleaned up by the router. + /// Defaults to 30 minutes if not set. + #[must_use] + pub fn with_stream_idle_timeout(mut self, timeout: std::time::Duration) -> Self { + self.stream_idle_timeout = Some(timeout); + self + } + /// Request a specific gateway instead of a random one. #[must_use] pub fn request_gateway(mut self, user_chosen_gateway: String) -> Self { @@ -352,6 +365,7 @@ where } /// Construct a [`DisconnectedMixnetClient`] from the setup specified. + #[allow(clippy::result_large_err)] pub fn build(self) -> Result> { let mut client = DisconnectedMixnetClient::new( self.config, @@ -375,6 +389,7 @@ where client.forget_me = self.forget_me; client.remember_me = self.remember_me; client.derivation_material = self.derivation_material; + client.stream_idle_timeout = self.stream_idle_timeout; Ok(client) } } @@ -443,6 +458,8 @@ where remember_me: RememberMe, /// The derivation material to use for the client keys, its up to the caller to save this for rederivation later derivation_material: Option, + + stream_idle_timeout: Option, } impl DisconnectedMixnetClient @@ -462,6 +479,7 @@ where /// Callers have the option of supplying further parameters to: /// - store persistent identities at a location on-disk, if desired; /// - use SOCKS5 mode + #[allow(clippy::result_large_err)] fn new( config: Config, socks5_config: Option, @@ -504,6 +522,7 @@ where forget_me, remember_me, derivation_material: None, + stream_idle_timeout: None, }) } @@ -906,6 +925,7 @@ where if self.socks5_config.is_some() { return Err(Error::Socks5Config { set: true }); } + let stream_idle_timeout = self.stream_idle_timeout; let (mut started_client, nym_address) = self.connect_to_mixnet_common().await?; let client_input = started_client.client_input.register_producer(); let mut client_output = started_client.client_output.register_consumer(); @@ -916,7 +936,7 @@ where let identity_keys = started_client.identity_keys.clone(); let reconstructed_receiver = client_output.register_receiver()?; - Ok(MixnetClient::new( + let mut client = MixnetClient::new( nym_address, identity_keys, client_input, @@ -928,7 +948,11 @@ where None, started_client.forget_me, started_client.remember_me, - )) + ); + if let Some(timeout) = stream_idle_timeout { + client.stream_idle_timeout = timeout; + } + Ok(client) } } diff --git a/sdk/rust/nym-sdk/src/mixnet/native_client.rs b/sdk/rust/nym-sdk/src/mixnet/native_client.rs index d9d19008d1..e2b9bb29f4 100644 --- a/sdk/rust/nym-sdk/src/mixnet/native_client.rs +++ b/sdk/rust/nym-sdk/src/mixnet/native_client.rs @@ -1,4 +1,6 @@ use crate::mixnet::client::MixnetClientBuilder; +use crate::mixnet::client::DEFAULT_NUMBER_OF_SURBS; +use crate::mixnet::stream::{MixnetListener, MixnetStream}; use crate::mixnet::traits::MixnetMessageSender; use crate::{Error, Result}; use async_trait::async_trait; @@ -21,8 +23,10 @@ use nym_task::connections::{ConnectionCommandSender, LaneQueueLengths}; use nym_task::ShutdownTracker; use nym_topology::{NymRouteProvider, NymTopology}; use std::pin::Pin; +use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; use std::task::{Context, Poll}; +use std::time::Duration; use tokio::sync::RwLockReadGuard; use tokio_util::sync::CancellationToken; @@ -47,7 +51,8 @@ pub struct MixnetClient { pub(crate) client_state: ClientState, /// A channel for messages arriving from the mixnet after they have been reconstructed. - pub(crate) reconstructed_receiver: ReconstructedMessagesReceiver, + /// Taken by the stream router on stream mode activation, `None` thereafter. + pub(crate) reconstructed_receiver: Option, /// A channel for sending stats event to be reported. pub(crate) stats_events_reporter: ClientStatsSender, @@ -56,10 +61,21 @@ pub struct MixnetClient { pub(crate) shutdown_handle: ShutdownTracker, pub(crate) packet_type: Option, - // internal state used for the `Stream` implementation + /// Internal state used for the `Stream` implementation _buffered: Vec, + pub(crate) forget_me: ForgetMe, pub(crate) remember_me: RememberMe, + + /// Set to `true` when the stream router is active, preventing + /// message-based functions from being used concurrently. + pub(crate) stream_mode: Arc, + + /// Opaque stream multiplexing state (lazily initialized by stream module). + pub(crate) streams: Option, + + /// How long a stream can be idle before the router cleans it up. + pub(crate) stream_idle_timeout: Duration, } impl MixnetClient { @@ -83,13 +99,16 @@ impl MixnetClient { client_input, client_output, client_state, - reconstructed_receiver, + reconstructed_receiver: Some(reconstructed_receiver), stats_events_reporter, shutdown_handle: task_handle, packet_type, _buffered: Vec::new(), forget_me, remember_me, + stream_mode: Arc::new(AtomicBool::new(false)), + streams: None, + stream_idle_timeout: super::stream::DEFAULT_STREAM_IDLE_TIMEOUT, } } @@ -156,6 +175,7 @@ impl MixnetClient { MixnetClientSender { client_input: self.client_input.clone(), packet_type: self.packet_type, + stream_mode: self.stream_mode.clone(), } } @@ -198,7 +218,11 @@ impl MixnetClient { /// Wait for messages from the mixnet pub async fn wait_for_messages(&mut self) -> Option> { - self.reconstructed_receiver.next().await + if self.stream_mode.load(Ordering::SeqCst) { + tracing::warn!("wait_for_messages() called after stream mode activated"); + return None; + } + self.reconstructed_receiver.as_mut()?.next().await } /// Provide a callback to execute on incoming messages from the mixnet. @@ -280,23 +304,66 @@ impl MixnetClient { } } } + + /// Open a stream to a remote peer. + /// + /// Returns a [`MixnetStream`] implementing `AsyncRead + AsyncWrite`. + /// `reply_surbs` controls how many reply SURBs are included with each + /// outbound message so the peer can reply. Defaults to 10 if `None`. + pub async fn open_stream( + &mut self, + recipient: Recipient, + reply_surbs: Option, + ) -> Result { + super::stream::open_stream( + self, + recipient, + reply_surbs.unwrap_or(DEFAULT_NUMBER_OF_SURBS), + ) + .await + } + + /// Create a listener that accepts inbound streams from remote peers. + /// + /// Can only be called once. + pub fn listener(&mut self) -> Result { + super::stream::listener(self) + } } -#[derive(Clone)] pub struct MixnetClientSender { client_input: ClientInput, packet_type: Option, + stream_mode: Arc, +} + +impl Clone for MixnetClientSender { + fn clone(&self) -> Self { + Self { + client_input: self.client_input.clone(), + packet_type: self.packet_type, + stream_mode: self.stream_mode.clone(), + } + } } impl Stream for MixnetClient { type Item = ReconstructedMessage; fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + if self.stream_mode.load(Ordering::SeqCst) { + tracing::warn!("Stream::poll_next() called after stream mode activated"); + return Poll::Ready(None); + } if let Some(next) = self._buffered.pop() { cx.waker().wake_by_ref(); return Poll::Ready(Some(next)); } - match ready!(Pin::new(&mut self.reconstructed_receiver).poll_next(cx)) { + let receiver = match self.reconstructed_receiver.as_mut() { + Some(rx) => rx, + None => return Poll::Ready(None), + }; + match ready!(Pin::new(receiver).poll_next(cx)) { None => Poll::Ready(None), Some(mut msgs) => { // the vector itself should never be empty @@ -327,6 +394,10 @@ impl MixnetMessageSender for MixnetClient { } async fn send(&self, message: InputMessage) -> Result<()> { + if self.stream_mode.load(Ordering::SeqCst) { + tracing::warn!("send() called after stream mode activated"); + return Err(Error::StreamModeActive); + } self.client_input .send(message) .await @@ -341,6 +412,10 @@ impl MixnetMessageSender for MixnetClientSender { } async fn send(&self, message: InputMessage) -> Result<()> { + if self.stream_mode.load(Ordering::SeqCst) { + tracing::warn!("send() called after stream mode activated"); + return Err(Error::StreamModeActive); + } self.client_input .send(message) .await diff --git a/sdk/rust/nym-sdk/src/mixnet/paths.rs b/sdk/rust/nym-sdk/src/mixnet/paths.rs index 5f8d60a7de..adcb5b9158 100644 --- a/sdk/rust/nym-sdk/src/mixnet/paths.rs +++ b/sdk/rust/nym-sdk/src/mixnet/paths.rs @@ -53,6 +53,7 @@ impl StoragePaths { /// /// This function will return an error if it is passed a path to an existing file instead of a /// directory. + #[allow(clippy::result_large_err)] pub fn new_from_dir>(dir: P) -> Result { let dir = dir.as_ref(); if dir.is_file() { diff --git a/sdk/rust/nym-sdk/src/mixnet/sink.rs b/sdk/rust/nym-sdk/src/mixnet/sink.rs index a3b8fd8ada..2161ce07e5 100644 --- a/sdk/rust/nym-sdk/src/mixnet/sink.rs +++ b/sdk/rust/nym-sdk/src/mixnet/sink.rs @@ -26,6 +26,7 @@ const SINK_BUFFER_SIZE_IN_MESSAGES: usize = 8; /// Traits that represents the ability to convert bytes into InputMessages that can be sent to the /// mixnet. This is typically used to set the destination and other sending parameters. pub trait MixnetMessageSinkTranslator: Unpin { + #[allow(clippy::result_large_err)] fn to_input_message(&self, bytes: &[u8]) -> Result; } diff --git a/sdk/rust/nym-sdk/src/mixnet/stream/mixnet_stream.rs b/sdk/rust/nym-sdk/src/mixnet/stream/mixnet_stream.rs new file mode 100644 index 0000000000..1d3358aeea --- /dev/null +++ b/sdk/rust/nym-sdk/src/mixnet/stream/mixnet_stream.rs @@ -0,0 +1,201 @@ +//! Per-stream handle implementing `AsyncRead + AsyncWrite`. + +use std::pin::Pin; +use std::task::{Context, Poll}; + +use bytes::BytesMut; +use futures::{ready, SinkExt}; +use tokio::io::{AsyncRead, AsyncWrite, ReadBuf}; +use tokio::sync::mpsc; + +use nym_client_core::client::base_client::ClientInput; +use nym_client_core::client::inbound_messages::InputMessage; +use nym_sphinx::addressing::clients::Recipient; +use nym_sphinx::anonymous_replies::requests::AnonymousSenderTag; +use nym_sphinx::params::PacketType; +use nym_task::connections::TransmissionLane; +use tokio_util::sync::PollSender; + +use super::protocol::{encode_stream_message, StreamId, StreamMessageType}; +use super::StreamMap; + +/// How to address outbound messages on this stream. +enum Destination { + /// We know the peer's Nym address. + Address { + recipient: Box, + reply_surbs: u32, + }, + /// We reply via the opener's anonymous sender tag. + Anonymous { sender_tag: AnonymousSenderTag }, +} + +/// A byte stream to a single remote Nym client. +/// +/// Provides `AsyncRead + AsyncWrite`. Created via +/// [`MixnetClient::open_stream`] (outbound) or +/// [`MixnetListener::accept`] (inbound). +pub struct MixnetStream { + id: StreamId, + destination: Destination, + sender: PollSender, + packet_type: Option, + streams: StreamMap, + + inbound_rx: mpsc::UnboundedReceiver>, + read_buf: BytesMut, + deregistered: bool, +} + +impl MixnetStream { + /// Create a stream we initiated to a known recipient. + pub(crate) fn new_outbound( + id: StreamId, + recipient: Recipient, + reply_surbs: u32, + client_input: ClientInput, + packet_type: Option, + streams: StreamMap, + inbound_rx: mpsc::UnboundedReceiver>, + ) -> Self { + let sender = PollSender::new(client_input.input_sender.clone()); + Self { + id, + destination: Destination::Address { + recipient: Box::new(recipient), + reply_surbs, + }, + sender, + packet_type, + streams, + inbound_rx, + read_buf: BytesMut::new(), + deregistered: false, + } + } + + /// Create a stream accepted from a remote peer's Open message. + pub(crate) fn new_inbound( + id: StreamId, + sender_tag: AnonymousSenderTag, + client_input: ClientInput, + packet_type: Option, + streams: StreamMap, + inbound_rx: mpsc::UnboundedReceiver>, + initial_data: Vec, + ) -> Self { + let mut read_buf = BytesMut::new(); + if !initial_data.is_empty() { + read_buf.extend_from_slice(&initial_data); + } + let sender = PollSender::new(client_input.input_sender.clone()); + Self { + id, + destination: Destination::Anonymous { sender_tag }, + sender, + packet_type, + streams, + inbound_rx, + read_buf, + deregistered: false, + } + } + + /// Return the unique identifier for this stream. + pub fn id(&self) -> StreamId { + self.id + } + + /// Wrap `data` in the appropriate `InputMessage` for this stream's destination. + fn make_input_message(&self, data: Vec) -> InputMessage { + match &self.destination { + Destination::Address { + recipient, + reply_surbs, + } => InputMessage::new_anonymous( + **recipient, + data, + *reply_surbs, + TransmissionLane::General, + self.packet_type, + ), + Destination::Anonymous { sender_tag } => InputMessage::new_reply( + *sender_tag, + data, + TransmissionLane::General, + self.packet_type, + ), + } + } +} + +impl Drop for MixnetStream { + fn drop(&mut self) { + if !self.deregistered { + self.streams.remove_background(self.id); + } + } +} + +impl AsyncRead for MixnetStream { + fn poll_read( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut ReadBuf, + ) -> Poll> { + // Drain spillover first + if !self.read_buf.is_empty() { + let n = std::cmp::min(buf.remaining(), self.read_buf.len()); + buf.put_slice(&self.read_buf.split_to(n)); + return Poll::Ready(Ok(())); + } + + match ready!(self.inbound_rx.poll_recv(cx)) { + Some(data) => { + let n = std::cmp::min(buf.remaining(), data.len()); + buf.put_slice(&data[..n]); + if n < data.len() { + self.read_buf.extend_from_slice(&data[n..]); + } + Poll::Ready(Ok(())) + } + None => Poll::Ready(Ok(())), // EOF + } + } +} + +impl AsyncWrite for MixnetStream { + fn poll_write( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + if buf.is_empty() { + return Poll::Ready(Ok(0)); + } + + ready!(self.sender.poll_ready_unpin(cx)) + .map_err(|_| std::io::Error::other("mixnet input channel closed"))?; + + let wire = encode_stream_message(&self.id, StreamMessageType::Data, buf); + let msg = self.make_input_message(wire); + + self.sender + .start_send_unpin(msg) + .map_err(|_| std::io::Error::other("failed to send stream message"))?; + + Poll::Ready(Ok(buf.len())) + } + + fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn poll_shutdown(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + if !self.deregistered { + self.streams.remove_background(self.id); + self.deregistered = true; + } + Poll::Ready(Ok(())) + } +} diff --git a/sdk/rust/nym-sdk/src/mixnet/stream/mod.rs b/sdk/rust/nym-sdk/src/mixnet/stream/mod.rs new file mode 100644 index 0000000000..2948837309 --- /dev/null +++ b/sdk/rust/nym-sdk/src/mixnet/stream/mod.rs @@ -0,0 +1,404 @@ +//! Stream multiplexing for `MixnetClient`. +//! +//! A [`MixnetStream`] is a byte channel (`AsyncRead + AsyncWrite`) to a +//! remote peer, identified by a [`StreamId`]. A single `MixnetClient` +//! can hold many streams to different peers concurrently. +//! +//! A background router task reads the client's `reconstructed_receiver`, +//! parses the stream header, and dispatches each payload to the right +//! stream's channel (or to the listener for `Open` messages). + +mod mixnet_stream; +mod protocol; + +pub use mixnet_stream::MixnetStream; +pub use protocol::StreamId; + +use std::collections::HashMap; +use std::sync::atomic::Ordering; +use std::sync::Arc; +use std::time::Duration; + +use tokio::time::Instant; + +use futures::StreamExt; +use tokio::sync::mpsc; +use tokio_util::sync::CancellationToken; +use tracing::{trace, warn}; + +use nym_client_core::client::inbound_messages::InputMessage; +use nym_client_core::client::received_buffer::ReconstructedMessagesReceiver; +use nym_sphinx::addressing::clients::Recipient; +use nym_sphinx::anonymous_replies::requests::AnonymousSenderTag; +use nym_task::connections::TransmissionLane; + +use protocol::{decode_stream_message, encode_stream_message, StreamMessageType}; + +use crate::mixnet::native_client::MixnetClient; +use crate::{Error, Result}; + +/// Default idle timeout before a stream is considered stale and cleaned up. +pub(crate) const DEFAULT_STREAM_IDLE_TIMEOUT: Duration = Duration::from_secs(30 * 60); + +/// Maximum interval between stale-stream checks. The actual check interval +/// is `min(idle_timeout, MAX_CLEANUP_INTERVAL)` so that short idle timeouts +/// are respected promptly rather than waiting up to 60 s for the next sweep. +const MAX_CLEANUP_INTERVAL: Duration = Duration::from_secs(10); + +/// Per-stream state stored in the routing table. +struct StreamEntry { + sender: mpsc::UnboundedSender>, + last_activity: Instant, +} + +/// The shared stream routing table. +/// +/// Wraps the map of active streams behind an async mutex with focused +/// methods so callers never touch the lock directly. +#[derive(Clone)] +pub(crate) struct StreamMap { + inner: Arc>>, +} + +impl StreamMap { + fn new() -> Self { + Self { + inner: Arc::new(tokio::sync::Mutex::new(HashMap::new())), + } + } + + /// Register a new stream, returning the receiver end of its data channel. + async fn register_stream(&self, stream_id: StreamId) -> mpsc::UnboundedReceiver> { + let (tx, rx) = mpsc::unbounded_channel(); + self.inner.lock().await.insert( + stream_id, + StreamEntry { + sender: tx, + last_activity: Instant::now(), + }, + ); + rx + } + + /// Remove a stream from the map. + async fn remove(&self, stream_id: &StreamId) { + self.inner.lock().await.remove(stream_id); + } + + /// Remove a stream without awaiting — for use in `Drop` and `poll_shutdown` + /// where we cannot `.await`. Spawns a lightweight background task. + fn remove_background(&self, stream_id: StreamId) { + let inner = self.inner.clone(); + tokio::spawn(async move { + inner.lock().await.remove(&stream_id); + }); + } + + /// Dispatch data to a stream's channel. Updates `last_activity` on + /// success. Removes the entry if the receiver has been dropped. + async fn send_to_stream(&self, stream_id: &StreamId, data: Vec) { + let mut map = self.inner.lock().await; + let should_remove = if let Some(entry) = map.get_mut(stream_id) { + if entry.sender.send(data).is_err() { + true + } else { + entry.last_activity = Instant::now(); + false + } + } else { + false + }; + if should_remove { + map.remove(stream_id); + } + } + + /// Remove streams that have been idle longer than `max_idle`. + async fn cleanup_stale(&self, max_idle: Duration) { + let now = Instant::now(); + let mut map = self.inner.lock().await; + map.retain(|id, entry| { + let stale = now.duration_since(entry.last_activity) >= max_idle; + if stale { + trace!("Cleaning up stale stream {id} (idle > {max_idle:?})"); + } + !stale + }); + } +} + +/// Delivered to the listener when a remote peer opens a new stream. +struct InboundOpen { + stream_id: StreamId, + sender_tag: Option, + initial_data: Vec, +} + +/// Owns the router task and the shared state for all streams on a client. +/// The router is a background task that reads reconstructed messages from the +/// mixnet, decodes the stream header, and dispatches each payload to the +/// correct stream's channel (or to the listener for new `Open` messages). +pub(crate) struct StreamState { + streams: StreamMap, + listener_rx: Option>, + shutdown: CancellationToken, + _router_handle: tokio::task::JoinHandle<()>, +} + +impl Drop for StreamState { + fn drop(&mut self) { + self.shutdown.cancel(); + } +} + +/// Accepts inbound streams opened by remote peers. +/// +/// Created via [`MixnetClient::listener`]. Each `accept()` returns a +/// `MixnetStream` ready for reading and writing. +pub struct MixnetListener { + inbound_rx: mpsc::UnboundedReceiver, + client_input: nym_client_core::client::base_client::ClientInput, + packet_type: Option, + streams: StreamMap, +} + +impl MixnetListener { + /// Wait for a remote peer to open a stream. + /// + /// Returns `None` if the router has shut down. + pub async fn accept(&mut self) -> Option { + loop { + let req = self.inbound_rx.recv().await?; + + let sender_tag = match req.sender_tag { + Some(tag) => tag, + None => { + warn!( + "Listener: Open for {} has no sender_tag, skipping", + req.stream_id + ); + continue; + } + }; + + let rx = self.streams.register_stream(req.stream_id).await; + + return Some(MixnetStream::new_inbound( + req.stream_id, + sender_tag, + self.client_input.clone(), + self.packet_type, + self.streams.clone(), + rx, + req.initial_data, + )); + } + } +} + +/// Background loop that demuxes incoming mixnet messages into per-stream channels. +async fn run_router( + mut reconstructed_rx: ReconstructedMessagesReceiver, + streams: StreamMap, + listener_tx: mpsc::UnboundedSender, + shutdown: CancellationToken, + idle_timeout: Duration, +) { + let check_every = std::cmp::min(idle_timeout, MAX_CLEANUP_INTERVAL); + let mut cleanup_interval = tokio::time::interval(check_every); + cleanup_interval.tick().await; // consume the immediate first tick + + loop { + let messages = tokio::select! { + _ = shutdown.cancelled() => break, + _ = cleanup_interval.tick() => { + streams.cleanup_stale(idle_timeout).await; + continue; + } + msg = reconstructed_rx.next() => match msg { + Some(messages) => messages, + None => break, + }, + }; + + for msg in messages { + let Some(frame) = decode_stream_message(&msg.message) else { + trace!( + "Router: non-stream message ({} bytes), dropping", + msg.message.len() + ); + continue; + }; + + let stream_id = frame.header.stream_id; + match frame.header.message_type { + StreamMessageType::Open => { + let _ = listener_tx.send(InboundOpen { + stream_id, + sender_tag: msg.sender_tag, + initial_data: frame.data.to_vec(), + }); + } + StreamMessageType::Data => { + streams + .send_to_stream(&stream_id, frame.data.to_vec()) + .await; + } // TODO: if we decide we need close logic add another enum member + } + } + } +} + +/// Lazily initialise the stream subsystem and router on first use. +fn ensure_init(client: &mut MixnetClient) -> Result<&mut StreamState> { + if client.streams.is_none() { + let real_rx = client + .reconstructed_receiver + .take() + .ok_or(Error::StreamInitFailure)?; + + // Set after take() succeeds so we don't leave the client in a + // broken state (stream_mode=true but no router) on failure. + client.stream_mode.store(true, Ordering::SeqCst); + + let streams = StreamMap::new(); + let (listener_tx, listener_rx) = mpsc::unbounded_channel(); + let shutdown = CancellationToken::new(); + + let router_handle = tokio::spawn(run_router( + real_rx, + streams.clone(), + listener_tx, + shutdown.clone(), + client.stream_idle_timeout, + )); + + client.streams = Some(StreamState { + streams, + listener_rx: Some(listener_rx), + shutdown, + _router_handle: router_handle, + }); + } + client.streams.as_mut().ok_or(Error::StreamInitFailure) +} + +/// Open a stream to a remote peer. +pub(crate) async fn open_stream( + client: &mut MixnetClient, + recipient: Recipient, + reply_surbs: u32, +) -> Result { + let streams = ensure_init(client)?.streams.clone(); + + let stream_id = StreamId::random(); + let rx = streams.register_stream(stream_id).await; + + // Send Open to the peer + let wire = encode_stream_message(&stream_id, StreamMessageType::Open, &[]); + let msg = InputMessage::new_anonymous( + recipient, + wire, + reply_surbs, + TransmissionLane::General, + client.packet_type, + ); + if (client.client_input.send(msg).await).is_err() { + streams.remove(&stream_id).await; + return Err(Error::MessageSendingFailure); + } + + Ok(MixnetStream::new_outbound( + stream_id, + recipient, + reply_surbs, + client.client_input.clone(), + client.packet_type, + streams, + rx, + )) +} + +/// Create a listener that accepts inbound streams. Can only be called once. +pub(crate) fn listener(client: &mut MixnetClient) -> Result { + let state = ensure_init(client)?; + let listener_rx = state + .listener_rx + .take() + .ok_or(Error::ListenerAlreadyTaken)?; + let streams = state.streams.clone(); + + Ok(MixnetListener { + inbound_rx: listener_rx, + client_input: client.client_input.clone(), + packet_type: client.packet_type, + streams, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test(start_paused = true)] + async fn cleanup_stale_removes_idle_streams() { + let map = StreamMap::new(); + let timeout = Duration::from_secs(10); + + // Register two streams + let _rx_a = map.register_stream(StreamId::random()).await; + let _rx_b = map.register_stream(StreamId::random()).await; + + // Advance time past the timeout + tokio::time::advance(timeout + Duration::from_secs(1)).await; + + // Register a fresh stream (should survive cleanup) + let id_c = StreamId::random(); + let _rx_c = map.register_stream(id_c).await; + + map.cleanup_stale(timeout).await; + + let inner = map.inner.lock().await; + assert_eq!(inner.len(), 1); + assert!(inner.contains_key(&id_c)); + } + + #[tokio::test(start_paused = true)] + async fn send_to_stream_updates_last_activity() { + let map = StreamMap::new(); + let timeout = Duration::from_secs(10); + let id = StreamId::random(); + + let _rx = map.register_stream(id).await; + + // Advance most of the way through the timeout + tokio::time::advance(Duration::from_secs(8)).await; + + // Activity on the stream resets its timer + map.send_to_stream(&id, vec![1, 2, 3]).await; + + // Advance past the original timeout, but only 5s since last activity + tokio::time::advance(Duration::from_secs(5)).await; + + map.cleanup_stale(timeout).await; + + // Stream should survive — last activity was 5s ago, not 13s + assert_eq!(map.inner.lock().await.len(), 1); + } + + #[tokio::test(start_paused = true)] + async fn cleanup_does_not_remove_active_streams() { + let map = StreamMap::new(); + let timeout = Duration::from_secs(10); + + let id = StreamId::random(); + let _rx = map.register_stream(id).await; + + // Advance less than the timeout + tokio::time::advance(Duration::from_secs(5)).await; + + map.cleanup_stale(timeout).await; + + assert_eq!(map.inner.lock().await.len(), 1); + } +} diff --git a/sdk/rust/nym-sdk/src/mixnet/stream/protocol.rs b/sdk/rust/nym-sdk/src/mixnet/stream/protocol.rs new file mode 100644 index 0000000000..0cd6891ee8 --- /dev/null +++ b/sdk/rust/nym-sdk/src/mixnet/stream/protocol.rs @@ -0,0 +1,198 @@ +//! Wire protocol for stream multiplexing. +//! +//! Every message between streams carries a fixed header prepended to +//! the payload inside the mixnet message body: +//! +//! ```text +//! [Version: 1 byte][StreamId: 8 bytes][MessageType: 1 byte][payload: N bytes] +//! ``` +//! +//! This header sits inside the sphinx packet payload. + +use std::fmt; + +/// Current stream protocol version. +pub const STREAM_PROTOCOL_VERSION: u8 = 1; + +/// Length of a StreamId in bytes (u64, big-endian). +pub const STREAM_ID_LEN: usize = 8; + +/// Total header length: Version (1) + StreamId (8) + MessageType (1). +pub const STREAM_HEADER_LEN: usize = 1 + STREAM_ID_LEN + 1; + +/// Identifies a stream within a MixnetClient. +#[derive(Clone, Copy, PartialEq, Eq, Hash)] +pub struct StreamId(u64); + +impl StreamId { + pub fn random() -> Self { + Self(rand::random::()) + } + + pub fn to_bytes(self) -> [u8; STREAM_ID_LEN] { + self.0.to_be_bytes() + } + + pub fn from_bytes(bytes: [u8; STREAM_ID_LEN]) -> Self { + Self(u64::from_be_bytes(bytes)) + } +} + +impl fmt::Debug for StreamId { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "StreamId({:#018x})", self.0) + } +} + +impl fmt::Display for StreamId { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + fmt::Debug::fmt(self, f) + } +} + +/// Message types within the stream protocol. +/// +/// Note: there is no Close variant. Without message sequencing, a close +/// message races ahead of in-flight data and arrives before the data is +/// reconstructed. Streams clean up locally via Drop. If ordered close/EOF +/// is needed in future, add sequencing + reorder buffering (see the +/// tcp_proxy's `MessageBuffer` for a working example). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[repr(u8)] +pub enum StreamMessageType { + /// Open a new stream. Payload is optional initial data. + Open = 0, + /// Data on an existing stream. + Data = 1, +} + +impl StreamMessageType { + pub fn from_byte(b: u8) -> Option { + match b { + 0 => Some(Self::Open), + 1 => Some(Self::Data), + _ => None, + } + } +} + +/// The fixed-size header prepended to every stream message. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct MixStreamHeader { + pub version: u8, + pub stream_id: StreamId, + pub message_type: StreamMessageType, +} + +/// A decoded stream frame: header + payload reference. +#[derive(Debug)] +pub struct MixStreamFrame<'a> { + pub header: MixStreamHeader, + pub data: &'a [u8], +} + +/// Encode a stream message: `[version][stream_id][msg_type][payload]`. +pub fn encode_stream_message( + id: &StreamId, + msg_type: StreamMessageType, + payload: &[u8], +) -> Vec { + let mut buf = Vec::with_capacity(STREAM_HEADER_LEN + payload.len()); + buf.push(STREAM_PROTOCOL_VERSION); + buf.extend_from_slice(&id.to_bytes()); + buf.push(msg_type as u8); + buf.extend_from_slice(payload); + buf +} + +/// Decode a stream message into a [`MixStreamFrame`]. +/// +/// Returns `None` if the buffer is too short, the version is unknown, +/// or the message type byte is invalid. +pub fn decode_stream_message(bytes: &[u8]) -> Option> { + if bytes.len() < STREAM_HEADER_LEN { + return None; + } + + let version = bytes[0]; + if version != STREAM_PROTOCOL_VERSION { + return None; + } + + let mut id_bytes = [0u8; STREAM_ID_LEN]; + id_bytes.copy_from_slice(&bytes[1..1 + STREAM_ID_LEN]); + let stream_id = StreamId::from_bytes(id_bytes); + + let message_type = StreamMessageType::from_byte(bytes[1 + STREAM_ID_LEN])?; + let data = &bytes[STREAM_HEADER_LEN..]; + + Some(MixStreamFrame { + header: MixStreamHeader { + version, + stream_id, + message_type, + }, + data, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn roundtrip() { + let id = StreamId::random(); + let payload = b"hello world"; + let encoded = encode_stream_message(&id, StreamMessageType::Data, payload); + let frame = decode_stream_message(&encoded).unwrap(); + assert_eq!(frame.header.version, STREAM_PROTOCOL_VERSION); + assert_eq!(frame.header.stream_id, id); + assert_eq!(frame.header.message_type, StreamMessageType::Data); + assert_eq!(frame.data, payload); + } + + #[test] + fn too_short() { + assert!(decode_stream_message(&[0u8; 5]).is_none()); + } + + #[test] + fn bad_version() { + let id = StreamId::random(); + let mut encoded = encode_stream_message(&id, StreamMessageType::Data, b"x"); + encoded[0] = 0xFF; + assert!(decode_stream_message(&encoded).is_none()); + } + + #[test] + fn bad_message_type() { + let mut buf = [0u8; STREAM_HEADER_LEN]; + buf[0] = STREAM_PROTOCOL_VERSION; + buf[1 + STREAM_ID_LEN] = 0xFF; + assert!(decode_stream_message(&buf).is_none()); + } + + #[test] + fn empty_payload() { + let id = StreamId::random(); + let encoded = encode_stream_message(&id, StreamMessageType::Open, &[]); + let frame = decode_stream_message(&encoded).unwrap(); + assert_eq!(frame.header.message_type, StreamMessageType::Open); + assert!(frame.data.is_empty()); + } + + #[test] + fn header_wire_format() { + let id = StreamId::from_bytes([0x00, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77]); + let encoded = encode_stream_message(&id, StreamMessageType::Open, &[0xAA]); + assert_eq!(encoded.len(), STREAM_HEADER_LEN + 1); + assert_eq!(encoded[0], STREAM_PROTOCOL_VERSION); + assert_eq!( + &encoded[1..9], + &[0x00, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77] + ); + assert_eq!(encoded[9], StreamMessageType::Open as u8); + assert_eq!(encoded[10], 0xAA); + } +}