diff --git a/sdk/rust/nym-sdk/src/lib.rs b/sdk/rust/nym-sdk/src/lib.rs index aff475d8cc..9e319e6f63 100644 --- a/sdk/rust/nym-sdk/src/lib.rs +++ b/sdk/rust/nym-sdk/src/lib.rs @@ -12,8 +12,6 @@ pub mod client_pool; pub mod ip_packet_client; pub mod ipr_wrapper; pub mod mixnet; -// stream_wrapper is superseded by ipr_wrapper (LP frame envelope over mixnet) -// pub mod stream_wrapper; pub mod tcp_proxy; pub use error::{Error, Result}; diff --git a/sdk/rust/nym-sdk/src/stream_wrapper.rs b/sdk/rust/nym-sdk/src/stream_wrapper.rs deleted file mode 100644 index 769d5ba87a..0000000000 --- a/sdk/rust/nym-sdk/src/stream_wrapper.rs +++ /dev/null @@ -1,52 +0,0 @@ -// Copyright 2024 - Nym Technologies SA -// SPDX-License-Identifier: GPL-2.0-only - -//! High-level streaming interface for the mixnet. -//! -//! # Basic Usage -//! ## Simple Send/Receive -//! -//! ```no_run -//! use nym_sdk::stream_wrapper::{MixSocket, MixStream, NetworkEnvironment}; -//! -//! #[tokio::main] -//! async fn main() -> Result<(), Box> { -//! let env = NetworkEnvironment::Mainnet; -//! -//! // Create listener (no peer) -//! let listener_socket = MixSocket::new(env.env_file_path()).await?; -//! let listener_address = *listener_socket.local_addr(); -//! let mut listener_stream = listener_socket.into_stream(); -//! -//! // Create sender connected to listener -//! let mut sender_stream = MixStream::connect(listener_address, env.env_file_path()).await?; -//! -//! // Sender initiates -//! sender_stream.send(b"Hello, Mixnet!").await?; -//! -//! // Listener receives and extracts SURB tag -//! let msg = listener_stream.recv().await?; -//! assert_eq!(msg.message, b"Hello, Mixnet!"); -//! -//! // Store SURB and reply anonymously -//! if let Some(surbs) = msg.sender_tag { -//! listener_stream.store_surb_tag(surbs); -//! listener_stream.send(b"Hello back!").await?; -//! } -//! -//! // Sender receives anonymous reply -//! let reply = sender_stream.recv().await?; -//! assert_eq!(reply.message, b"Hello back!"); -//! -//! Ok(()) -//! } -//! ``` -//! - -mod mixnet_stream_wrapper; -mod mixnet_stream_wrapper_ipr; -mod network_env; - -pub use mixnet_stream_wrapper::{MixSocket, MixStream, MixStreamReader, MixStreamWriter}; -pub use mixnet_stream_wrapper_ipr::{IpMixStream, IpMixStreamReader, IpMixStreamWriter}; -pub use network_env::NetworkEnvironment; diff --git a/sdk/rust/nym-sdk/src/stream_wrapper/mixnet_stream_wrapper.rs b/sdk/rust/nym-sdk/src/stream_wrapper/mixnet_stream_wrapper.rs deleted file mode 100644 index 5c4b3bd770..0000000000 --- a/sdk/rust/nym-sdk/src/stream_wrapper/mixnet_stream_wrapper.rs +++ /dev/null @@ -1,1057 +0,0 @@ -// Copyright 2024 - Nym Technologies SA -// SPDX-License-Identifier: GPL-3.0-only - -use crate::mixnet::InputMessage; -use crate::mixnet::MixnetMessageSender; -use crate::mixnet::{MixnetClient, MixnetClientSender, Recipient}; -use crate::Error; -use bytes::BytesMut; -use nym_client_core::client::inbound_messages::InputMessageCodec; -use nym_network_defaults::setup_env; -use nym_sphinx::anonymous_replies::requests::AnonymousSenderTag; -use nym_sphinx::receiver::{ReconstructedMessage, ReconstructedMessageCodec}; -use std::io; -use std::path::PathBuf; -use std::pin::Pin; -use std::task::{Context, Poll}; -use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt, ReadBuf}; -use tokio::sync::oneshot; -use tokio_util::codec::{Encoder, Framed, FramedRead, FramedWrite}; -use tracing::{debug, info, warn}; - -impl MixnetClient { - /// Send data to a recipient with reply SURBs. - /// - /// This is a high-level method that abstracts away codec details. - /// - /// # Arguments - /// * `recipient` - The Nym address to send to - /// * `data` - The message payload - /// * `reply_surbs` - Number of Single Use Reply Blocks to include. If `None` passed defaults to 10. - pub async fn send_to( - &mut self, - recipient: &Recipient, - data: &[u8], - reply_surbs: Option, - ) -> Result<(), Error> { - let msg = InputMessage::Anonymous { - recipient: *recipient, - data: data.to_vec(), - reply_surbs: reply_surbs.unwrap_or(10), - lane: nym_task::connections::TransmissionLane::General, - max_retransmissions: Some(5), - }; - MixnetMessageSender::send(self, msg).await - } - - /// Send a reply using a previously received SURB tag. - /// - /// This enables anonymous replies without knowing the original sender's address. - /// - /// # Arguments - /// * `recipient_tag` - The SURB tag from a received message - /// * `data` - The reply payload - pub async fn send_stream_reply( - &mut self, - recipient_tag: AnonymousSenderTag, - data: &[u8], - ) -> Result<(), Error> { - let msg = InputMessage::Reply { - recipient_tag, - data: data.to_vec(), - lane: nym_task::connections::TransmissionLane::General, - max_retransmissions: Some(5), - }; - MixnetMessageSender::send(self, msg).await - } - - /// Receive the next message, awaiting until one arrives. - /// - /// This method blocks until a complete message is received and decoded. - /// Uses framed reading internally for efficient message handling. - /// - /// # Returns - /// A `ReconstructedMessage` containing the data and optional sender tag - pub async fn recv(&mut self) -> Result { - use futures::StreamExt; - - let mut framed = self.framed_read(); - framed - .next() - .await - .ok_or_else(|| { - Error::IoError(io::Error::new( - io::ErrorKind::UnexpectedEof, - "Connection closed", - )) - })? - .map_err(Error::from) - } - - /// Try to receive a message without blocking. - /// - /// Returns `None` immediately if no complete message is available. - /// - /// # Returns - /// `Some(ReconstructedMessage)` if a message is available, `None` otherwise - pub async fn try_recv(&mut self) -> Result, Error> { - Ok(self.wait_for_messages().await.and_then(|mut msgs| { - if msgs.is_empty() { - None - } else { - Some(msgs.remove(0)) - } - })) - } - - /// Convert into a framed reader for stream-like message handling. - /// - /// Returns a `FramedRead` that automatically decodes `ReconstructedMessage`s. - /// - /// # Example - /// ```no_run - /// use futures::StreamExt; - /// use nym_sdk::mixnet::MixnetClient; - /// - /// #[tokio::main] - /// async fn main() -> Result<(), Box> { - /// let mut client = MixnetClient::connect_new().await?; - /// let mut framed = client.into_framed_read(); - /// while let Some(msg) = framed.next().await { - /// let msg = msg?; - /// println!("Received: {:?}", msg.message); - /// } - /// Ok(()) - /// } - /// ``` - pub fn into_framed_read(self) -> FramedRead { - FramedRead::new(self, ReconstructedMessageCodec {}) - } - - /// Create a framed reader without consuming self. - /// - /// Useful when you need to keep the client for other operations. - pub fn framed_read(&mut self) -> FramedRead<&mut Self, ReconstructedMessageCodec> { - FramedRead::new(self, ReconstructedMessageCodec {}) - } -} - -impl MixnetClientSender { - /// Send data to a recipient with optional reply SURBs. - /// - /// This is a high-level method that abstracts away codec details. - /// Identical to `MixnetClient::send_to` but for the split sender. - /// - /// # Arguments - /// * `recipient` - The Nym address to send to - /// * `data` - The message payload - /// * `reply_surbs` - Number of Single Use Reply Blocks to include. If `None` passed defaults to 10. - pub async fn send_to( - &mut self, - recipient: &Recipient, - data: &[u8], - reply_surbs: Option, - ) -> Result<(), Error> { - let msg = InputMessage::Anonymous { - recipient: *recipient, - data: data.to_vec(), - reply_surbs: reply_surbs.unwrap_or(10), - lane: nym_task::connections::TransmissionLane::General, - max_retransmissions: Some(5), - }; - MixnetMessageSender::send(self, msg).await - } - - /// Send a reply using a previously received SURB tag. - /// - /// # Arguments - /// * `recipient_tag` - The SURB tag from a received message - /// * `data` - The reply payload - pub async fn send_stream_reply( - &mut self, - recipient_tag: AnonymousSenderTag, - data: &[u8], - ) -> Result<(), Error> { - let msg = InputMessage::Reply { - recipient_tag, - data: data.to_vec(), - lane: nym_task::connections::TransmissionLane::General, - max_retransmissions: Some(5), - }; - MixnetMessageSender::send(self, msg).await - } -} - -/// A mixnet socket, similar to `TcpSocket`. -/// -/// Provides a high-level interface for creating mixnet connections -/// without dealing with codecs or low-level message handling. -pub struct MixSocket { - pub inner: MixnetClient, -} - -impl MixSocket { - /// Create a new socket connected to the mixnet. - /// - /// Initializes a new mixnet client and prepares it for connections. - /// # Arguments - /// * `env` - The environment to use. - pub async fn new(env: PathBuf) -> Result { - debug!("Loading env file: {:?}", env); - setup_env(Some(env)); - let inner = MixnetClient::connect_new().await?; - Ok(MixSocket { inner }) - } - - /// Connect to a specific peer and return a `MixStream`. - /// - /// Similar to `TcpSocket::connect`, establishes a connection - /// to a peer identified by their Nym address. - /// - /// # Arguments - /// * `recipient` - The Nym address of the peer to connect to - pub async fn connect(self, recipient: Recipient) -> Result { - Ok(MixStream { - client: self.inner, - peer: Some(recipient), - peer_surb_tag: None, - }) - } - - /// Convert socket into a listening stream without a specific peer. - /// - /// Creates a stream that can receive messages from anyone and reply - /// anonymously using SURBs. This is the mixnet equivalent of a listening socket. - /// - /// # Example - /// ```no_run - /// use nym_sdk::stream_wrapper::MixSocket; - /// use std::path::PathBuf; - /// - /// #[tokio::main] - /// async fn main() -> Result<(), Box> { - /// let env = PathBuf::from("path/to/env"); - /// let socket = MixSocket::new(env).await?; - /// let mut listener = socket.into_stream(); - /// - /// // Receive from anyone - /// let msg = listener.recv().await?; - /// - /// // Reply anonymously using SURB - /// if let Some(surb) = msg.sender_tag { - /// listener.store_surb_tag(surb); - /// listener.send(b"Reply").await?; - /// } - /// Ok(()) - /// } - /// ``` - pub fn into_stream(self) -> MixStream { - MixStream { - client: self.inner, - peer: None, - peer_surb_tag: None, - } - } - - /// Get our Nym address (like `TcpSocket::local_addr`). - pub fn local_addr(&self) -> &Recipient { - self.inner.nym_address() - } - - /// Get a reference to the underlying `MixnetClient`. - pub fn get_ref(&self) -> &MixnetClient { - &self.inner - } - - /// Get a mutable reference to the underlying `MixnetClient`. - pub fn get_mut(&mut self) -> &mut MixnetClient { - &mut self.inner - } - - /// Consume the socket and return the underlying `MixnetClient`. - pub fn into_inner(self) -> MixnetClient { - self.inner - } -} - -/// A mixnet stream, similar to `TcpStream`. -/// -/// Provides bidirectional communication with a peer over the mixnet. -/// Can operate in two modes: -/// - Connected mode: Has a specific peer address -/// - Listening mode: No peer, receives from anyone and replies via SURBs -pub struct MixStream { - pub client: MixnetClient, - peer: Option, - peer_surb_tag: Option, -} - -impl MixStream { - /// Create a `MixStream` from an optional socket and optional peer. - /// - /// If no socket is provided, creates a new one automatically. - /// If no peer is provided, creates a listening stream. - /// - /// # Arguments - /// * `socket` - Optional existing socket to use - /// * `peer` - Optional Nym address to connect to (None = listening mode) - /// * `env` - Optional environment to use. Required if not passing in an existing `socket`. - pub async fn new( - socket: Option, - peer: Option, - env: Option, - ) -> Result { - let client = match socket { - Some(socket) => socket.into_inner(), - None => { - if env.is_none() { - return Err(Error::MissingStreamConfig); - } - setup_env(env); - MixnetClient::connect_new().await.unwrap() - } - }; - Ok(Self { - client, - peer, - peer_surb_tag: None, - }) - } - - /// Create a listening stream that receives from anyone. - /// - /// This stream has no specific peer and relies on SURBs for replies. - /// For server-like applications that respond to anonymous requests. - /// - /// # Arguments - /// * `env` - The environment to use. Defaults to Mainnet if `None`. TODO MAKE PATH AND MAKE MORE CLEAR IN DOCS - /// - /// # Example - /// ```no_run - /// use nym_sdk::stream_wrapper::MixStream; - /// use std::path::PathBuf; - /// - /// #[tokio::main] - /// async fn main() -> Result<(), Box> { - /// let env = PathBuf::from("path/to/env"); - /// let mut listener = MixStream::listen(env).await?; - /// let msg = listener.recv().await?; - /// if let Some(surb) = msg.sender_tag { - /// listener.store_surb_tag(surb); - /// listener.send(b"Response").await?; - /// } - /// Ok(()) - /// } - /// ``` - pub async fn listen(env: PathBuf) -> Result { - debug!("Loading env file: {:?}", env); - setup_env(Some(env)); - let client = MixnetClient::connect_new().await?; - Ok(Self { - client, - peer: None, - peer_surb_tag: None, - }) - } - - /// Create a new `MixStream` and connect to a peer (like `TcpStream::connect`). - /// - /// This is a convenience method that creates both the socket and stream. - /// - /// # Arguments - /// * `peer` - The Nym address to connect to - /// * `env` - The environment to use. - pub async fn connect(peer: Recipient, env: PathBuf) -> Result { - debug!("Loading env file: {:?}", env); - setup_env(Some(env.clone())); - let socket = MixSocket::new(env).await?; - socket.connect(peer).await - } - - /// Get the peer's Nym address. - /// - /// Returns `None` if this is a listening stream. - pub fn peer_addr(&self) -> Option<&Recipient> { - self.peer.as_ref() - } - - /// Get our local Nym address (like `TcpStream::local_addr`). - pub fn local_addr(&self) -> &Recipient { - self.client.nym_address() - } - - /// Store a SURB tag for sending anonymous replies. - /// - /// SURB tags are typically extracted from received messages and enable - /// replying without knowing the original sender's address. - /// - /// # Arguments - /// * `surbs` - The sender tag from a received message - pub fn store_surb_tag(&mut self, surbs: AnonymousSenderTag) { - self.peer_surb_tag = Some(surbs); - } - - /// Get the currently stored SURB tag, if any. - pub fn surbs(&self) -> Option { - self.peer_surb_tag - } - - /// Send data to the peer or via SURB. - /// - /// Behavior depends on stream state: - /// - If SURB tag is stored: Uses SURBs from denoted bucket for anonymous reply - /// - If peer is set: Sends to peer with new reply SURBs - /// - If neither: Returns error - /// - /// # Arguments - /// * `data` - The message payload to send - pub async fn send(&mut self, data: &[u8]) -> Result<(), Error> { - match (self.peer_surb_tag, self.peer) { - (Some(tag), _) => { - // Have SURB tag - use it for anonymous reply - self.client.send_stream_reply(tag, data).await - } - (None, Some(peer)) => { - // Have peer - send with default # of SURBs - self.client.send_to(&peer, data, None).await - } - (None, None) => { - // No peer, no SURB - can't send - Err(Error::MixStreamNoPeerOrSurb) - } - } - } - - /// Receive the next message, awaiting until one arrives. - /// - /// Blocks until a complete message is received. - /// - /// # Returns - /// A `ReconstructedMessage` containing the data and optional sender tag - pub async fn recv(&mut self) -> Result { - self.client.recv().await - } - - /// Try to receive a message without blocking. - /// - /// Returns `None` immediately if no message is available. - /// Low-level method primarily for debugging. - /// - /// # Returns - /// `Some(ReconstructedMessage)` if available, `None` otherwise - pub async fn try_recv(&mut self) -> Result, Error> { - self.client.try_recv().await - } - - /// Split the stream for concurrent read/write operations (like `TcpStream::split`). - /// - /// Returns separate reader and writer halves that can be used concurrently. - /// The reader and writer share SURB tags via an internal channel. - pub fn split(self) -> (MixStreamReader, MixStreamWriter) { - debug!("Splitting MixStream"); - let sender = self.client.split_sender(); - debug!("Split MixStream into Reader and Writer"); - let (surb_tx, surb_rx) = oneshot::channel(); - ( - MixStreamReader { - client: self.client, - peer: self.peer, - peer_surb_tag: self.peer_surb_tag, - surb_tx: Some(surb_tx), - }, - MixStreamWriter { - sender, - peer: self.peer, - peer_surb_tag: self.peer_surb_tag, - surb_rx: Some(surb_rx), - }, - ) - } - - /// Convert into a framed stream for bidirectional message handling. - /// - /// Returns a `Framed` that automatically encodes/decodes messages. - /// - /// # Example - /// ```no_run - /// use futures::{SinkExt, StreamExt}; - /// use nym_sdk::mixnet::{InputMessage, Recipient}; - /// use nym_sdk::stream_wrapper::MixStream; - /// use std::path::PathBuf; - /// - /// #[tokio::main] - /// async fn main() -> Result<(), Box> { - /// let env = PathBuf::from("path/to/env"); - /// let peer = Recipient::try_from_base58_string("recipient_address")?; - /// let stream = MixStream::connect(peer, env).await?; - /// let mut framed = stream.into_framed(); - /// - /// if let Some(msg) = framed.next().await { - /// let msg = msg?; - /// println!("Received: {:?}", msg.message); - /// } - /// Ok(()) - /// } - /// ``` - pub fn into_framed(self) -> Framed { - Framed::new(self.client, ReconstructedMessageCodec {}) - } - - /// Convert into a framed reader for receiving messages. - /// - /// Returns a `FramedRead` that automatically decodes `ReconstructedMessage`s. - pub fn into_framed_read(self) -> FramedRead { - FramedRead::new(self.client, ReconstructedMessageCodec {}) - } - - /// Create a framed reader without consuming self. - pub fn framed_read(&mut self) -> FramedRead<&mut MixnetClient, ReconstructedMessageCodec> { - FramedRead::new(&mut self.client, ReconstructedMessageCodec {}) - } - - /// Write bytes using the codec. - /// - /// This is a low-level method for debugging. - /// Prefer using `send()` for normal operations. - /// - /// # Arguments - /// * `data` - Raw bytes to encode and send - pub async fn write_bytes(&mut self, data: &[u8]) -> Result<(), Error> { - let input_message = match (self.peer_surb_tag, self.peer) { - (Some(tag), _) => { - info!("Writing {} bytes, sending with SURBs", data.len()); - InputMessage::Reply { - recipient_tag: tag, - data: data.to_owned(), - lane: nym_task::connections::TransmissionLane::General, - max_retransmissions: Some(5), - } - } - (None, Some(peer)) => { - info!("Writing {} bytes", data.len()); - InputMessage::Anonymous { - recipient: peer, - data: data.to_owned(), - reply_surbs: 10, - lane: nym_task::connections::TransmissionLane::General, - max_retransmissions: Some(5), - } - } - (None, None) => { - return Err(Error::MixStreamNoPeerOrSurb); - } - }; - - let mut codec = InputMessageCodec {}; - let mut serialized_bytes = BytesMut::new(); - codec.encode(input_message, &mut serialized_bytes)?; - self.write_all(&serialized_bytes).await?; - debug!("Wrote serialized bytes"); - self.flush().await?; - debug!("Flushed"); - - Ok(()) - } - - /// Wait for messages using the low-level interface. - /// - /// This is a method for debugging only. - /// Prefer using `recv()` for normal operations. - pub async fn wait_for_messages(&mut self) -> Option> { - self.client.wait_for_messages().await - } - - /// Disconnect from the mixnet (like `TcpStream::shutdown`). - /// - /// Gracefully closes the connection. - pub async fn shutdown(self) -> Result<(), Error> { - debug!("Disconnecting"); - self.client.disconnect().await; - debug!("Disconnected"); - Ok(()) - } -} - -impl AsyncRead for MixStream { - fn poll_read( - mut self: Pin<&mut Self>, - cx: &mut Context<'_>, - buf: &mut ReadBuf<'_>, - ) -> Poll> { - Pin::new(&mut self.client).poll_read(cx, buf) - } -} - -impl AsyncWrite for MixStream { - fn poll_write( - mut self: Pin<&mut Self>, - cx: &mut Context<'_>, - buf: &[u8], - ) -> Poll> { - Pin::new(&mut self.client).poll_write(cx, buf) - } - - fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - Pin::new(&mut self.client).poll_flush(cx) - } - - fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - Pin::new(&mut self.client).poll_shutdown(cx) - } -} - -/// Read half of a split `MixStream`. -/// -/// Provides read-only access to the stream with automatic SURB tag handling. -pub struct MixStreamReader { - client: MixnetClient, - peer: Option, - peer_surb_tag: Option, - surb_tx: Option>, -} - -impl MixStreamReader { - /// Get the peer's Nym address. - pub fn peer_addr(&self) -> Option<&Recipient> { - self.peer.as_ref() - } - - /// Get our local Nym address. - pub fn local_addr(&self) -> &Recipient { - self.client.nym_address() - } - - /// Store a SURB tag and forward it to the writer half. - /// - /// Automatically sends the tag to the paired `MixStreamWriter` via - /// an internal channel for seamless anonymous replies. - /// - /// # Arguments - /// * `surbs` - The sender tag to store and forward - pub fn store_surb_tag(&mut self, surbs: AnonymousSenderTag) { - self.peer_surb_tag = Some(surbs); - if let Some(tx) = self.surb_tx.take() { - match tx.send(surbs) { - Ok(()) => debug!("Sent SURBs to MixStreamWriter"), - Err(e) => warn!("Could not send SURBs to MixStreamWriter with err: {}", e), - } - } - } - - /// Get the currently stored SURB tag, if any. - pub fn surbs(&self) -> Option { - self.peer_surb_tag - } - - /// Receive the next message, awaiting until one arrives. - /// - /// Automatically extracts and stores - /// SURB tags from received messages. - /// - /// # Returns - /// A `ReconstructedMessage` containing the data and optional sender tag - pub async fn recv(&mut self) -> Result { - use futures::StreamExt; - - let mut framed = self.framed(); - let msg = framed - .next() - .await - .ok_or_else(|| { - Error::IoError(io::Error::new( - io::ErrorKind::UnexpectedEof, - "Connection closed", - )) - })? - .map_err(Error::from)?; - - // Auto-store SURB tag if present - if let Some(surb) = msg.sender_tag { - self.store_surb_tag(surb); - } - - Ok(msg) - } - - /// Try to receive a message without blocking. - /// - /// Useful for debugging. Automatically stores - /// SURB tags if present. Prefer `recv()` in - /// normal use. - /// - /// # Returns - /// `Some(ReconstructedMessage)` if available, `None` otherwise - pub async fn try_recv(&mut self) -> Result, Error> { - let msg_opt = self.client.wait_for_messages().await.and_then(|mut msgs| { - if msgs.is_empty() { - None - } else { - Some(msgs.remove(0)) - } - }); - - if let Some(ref msg) = msg_opt { - if let Some(surb) = msg.sender_tag { - self.store_surb_tag(surb); - } - } - - Ok(msg_opt) - } - - /// Convert into a framed reader for stream-like message handling. - /// - /// Returns a `FramedRead` that automatically decodes `ReconstructedMessage`s - /// and handles SURB tag extraction. - /// - /// # Example - /// ```no_run - /// use futures::StreamExt; - /// use nym_sdk::stream_wrapper::MixStreamReader; - /// - /// async fn example(reader: MixStreamReader) -> Result<(), Box> { - /// let mut framed = reader.into_framed(); - /// while let Some(msg) = framed.next().await { - /// let msg = msg?; - /// if let Some(_surb) = msg.sender_tag { - /// // SURB automatically stored - /// } - /// } - /// Ok(()) - /// } - /// ``` - pub fn into_framed(self) -> FramedRead { - FramedRead::new(self.client, ReconstructedMessageCodec {}) - } - - /// Create a framed reader without consuming self. - pub fn framed(&mut self) -> FramedRead<&mut MixnetClient, ReconstructedMessageCodec> { - FramedRead::new(&mut self.client, ReconstructedMessageCodec {}) - } -} - -impl AsyncRead for MixStreamReader { - fn poll_read( - mut self: Pin<&mut Self>, - cx: &mut Context<'_>, - buf: &mut ReadBuf<'_>, - ) -> Poll> { - Pin::new(&mut self.client).poll_read(cx, buf) - } -} - -/// Write half of a split `MixStream`. -/// -/// Provides write-only access to the stream with automatic SURB tag -/// synchronization from the reader half. -pub struct MixStreamWriter { - sender: MixnetClientSender, - peer: Option, - peer_surb_tag: Option, - surb_rx: Option>, -} - -impl MixStreamWriter { - /// Send data to the connected peer or via SURB. - /// - /// Automatically checks for SURB tags - /// received by the paired reader and uses them for replies. - /// - /// # Arguments - /// * `data` - The message payload to send - pub async fn send(&mut self, data: &[u8]) -> Result<(), Error> { - // Check for SURB updates from reader - if self.peer_surb_tag.is_none() { - if let Some(rx) = self.surb_rx.as_mut() { - if let Ok(surbs) = rx.try_recv() { - self.peer_surb_tag = Some(surbs); - } - } - } - - match (self.peer_surb_tag, self.peer) { - (Some(tag), _) => { - // Have SURB tag - use it for anonymous reply - self.sender.send_stream_reply(tag, data).await - } - (None, Some(peer)) => { - // Have peer - send with SURBs - self.sender.send_to(&peer, data, Some(10)).await - } - (None, None) => { - // No peer, no SURB tag - can't send - Err(Error::MixStreamNoPeerOrSurb) - } - } - } - - /// Write bytes using the codec. - /// - /// Used for debugging and compatibility. - /// Prefer using `send()` for normal operations. - /// - /// # Arguments - /// * `data` - Raw bytes to encode and send - pub async fn write_bytes(&mut self, data: &[u8]) -> Result<(), Error> { - if self.peer_surb_tag.is_none() { - if let Some(rx) = self.surb_rx.as_mut() { - if let Ok(surbs) = rx.try_recv() { - self.peer_surb_tag = Some(surbs); - } - } - } - - let input_message = match (self.peer_surb_tag, self.peer) { - (Some(tag), _) => InputMessage::Reply { - recipient_tag: tag, - data: data.to_owned(), - lane: nym_task::connections::TransmissionLane::General, - max_retransmissions: Some(5), - }, - (None, Some(peer)) => InputMessage::Anonymous { - recipient: peer, - data: data.to_owned(), - reply_surbs: 10, - lane: nym_task::connections::TransmissionLane::General, - max_retransmissions: Some(5), - }, - (None, None) => { - return Err(Error::MixStreamNoPeerOrSurb); - } - }; - - let mut codec = InputMessageCodec {}; - let mut serialized_bytes = BytesMut::new(); - codec.encode(input_message, &mut serialized_bytes)?; - - self.write_all(&serialized_bytes).await?; - debug!("Wrote serialized bytes"); - self.flush().await?; - debug!("Flushed"); - - Ok(()) - } - - /// Convert into a framed writer for sending messages. - /// - /// Returns a `FramedWrite` that automatically encodes `InputMessage`s. - pub fn into_framed(self) -> FramedWrite { - FramedWrite::new(self.sender, InputMessageCodec {}) - } - - /// Create a framed writer without consuming self. - pub fn framed(&mut self) -> FramedWrite<&mut MixnetClientSender, InputMessageCodec> { - FramedWrite::new(&mut self.sender, InputMessageCodec {}) - } -} - -impl AsyncWrite for MixStreamWriter { - fn poll_write( - mut self: Pin<&mut Self>, - cx: &mut Context<'_>, - buf: &[u8], - ) -> Poll> { - Pin::new(&mut self.sender).poll_write(cx, buf) - } - - fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - Pin::new(&mut self.sender).poll_flush(cx) - } - - fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - Pin::new(&mut self.sender).poll_shutdown(cx) - } -} - -#[cfg(test)] -mod tests { - use super::*; - use crate::stream_wrapper::network_env::NetworkEnvironment; - use futures::StreamExt; - - #[tokio::test] - async fn simple_send_recv() -> Result<(), Box> { - let env = NetworkEnvironment::Mainnet; - - // Create listener (no peer) - let listener_socket = MixSocket::new(env.env_file_path()).await?; - let listener_address = *listener_socket.local_addr(); - let mut listener_stream = listener_socket.into_stream(); - - // Create sender connected to listener - let mut sender_stream = MixStream::connect(listener_address, env.env_file_path()).await?; - - // Sender initiates - sender_stream.send(b"Hello, Mixnet!").await?; - info!("Sent initial message"); - - // Listener receives and extracts SURB tag - let msg = - tokio::time::timeout(tokio::time::Duration::from_secs(30), listener_stream.recv()) - .await??; - - assert_eq!(msg.message, b"Hello, Mixnet!"); - info!("Received initial message"); - - // Store SURB and reply anonymously - if let Some(surbs) = msg.sender_tag { - listener_stream.store_surb_tag(surbs); - listener_stream.send(b"Hello back!").await?; - info!("Sent reply using SURB"); - } - - // Sender receives anonymous reply - let reply = - tokio::time::timeout(tokio::time::Duration::from_secs(30), sender_stream.recv()) - .await??; - - assert_eq!(reply.message, b"Hello back!"); - info!("Received SURB reply"); - - Ok(()) - } - - #[tokio::test] - async fn framed() -> Result<(), Box> { - let env = NetworkEnvironment::Mainnet; - - let receiver_socket = MixSocket::new(env.env_file_path()).await?; - let receiver_address = *receiver_socket.local_addr(); - let mut receiver_stream = receiver_socket.into_stream(); - - let sender_socket = MixSocket::new(env.env_file_path()).await?; - let sender_stream = sender_socket.connect(receiver_address).await?; - - let (sender_reader, mut sender_writer) = sender_stream.split(); - let mut sender_framed = sender_reader.into_framed(); - - sender_writer.send(b"Hello via framed!").await?; - info!("Sent message"); - - let msg = - tokio::time::timeout(tokio::time::Duration::from_secs(30), receiver_stream.recv()) - .await??; - - assert_eq!(msg.message, b"Hello via framed!"); - - if let Some(surb) = msg.sender_tag { - receiver_stream.store_surb_tag(surb); - receiver_stream.send(b"Reply via framed!").await?; - info!("Sent reply"); - } - - let reply = - tokio::time::timeout(tokio::time::Duration::from_secs(30), sender_framed.next()) - .await? - .ok_or("No reply received")??; - - assert_eq!(reply.message, b"Reply via framed!"); - - Ok(()) - } - - #[tokio::test] - async fn split_concurrent() -> Result<(), Box> { - let env = NetworkEnvironment::Mainnet; - - let sender_socket = MixSocket::new(env.env_file_path()).await?; - let receiver_socket = MixSocket::new(env.env_file_path()).await?; - let receiver_address = *receiver_socket.local_addr(); - - // Passing no env var to MixStream since we're passing a socket that is already configured to use Mainnet - let sender_stream = - MixStream::new(Some(sender_socket), Some(receiver_address), None).await?; - let receiver_stream = receiver_socket.into_stream(); - - let (mut sender_reader, mut sender_writer) = sender_stream.split(); - let (mut receiver_reader, mut receiver_writer) = receiver_stream.split(); - - let message_back_and_forth = 5; - - let sender_task = tokio::spawn(async move { - for i in 0..message_back_and_forth { - let msg = format!("Message {}", i); - sender_writer.send(msg.as_bytes()).await?; - info!("Sent message {}", i); - tokio::time::sleep(tokio::time::Duration::from_millis(200)).await; - } - Ok::<_, Error>(sender_writer) - }); - - let receiver_task = tokio::spawn(async move { - let mut received_count = 0; - while received_count < 5 { - match tokio::time::timeout( - tokio::time::Duration::from_secs(15), - receiver_reader.recv(), - ) - .await - { - Ok(Ok(msg)) => { - info!("Received: {}", String::from_utf8_lossy(&msg.message)); - received_count += 1; - let reply = format!("Reply {}", received_count); - receiver_writer.send(reply.as_bytes()).await?; - info!("Sent reply {}", received_count); - tokio::time::sleep(tokio::time::Duration::from_millis(200)).await; - } - Err(_) => { - warn!("Timeout waiting for message {}", received_count); - break; - } - Ok(Err(e)) => { - warn!("Error receiving message: {}", e); - break; - } - } - } - - Ok::<_, Error>((receiver_reader, receiver_writer, received_count)) - }); - - let reply_reader_task = tokio::spawn(async move { - let mut reply_count = 0; - while reply_count < message_back_and_forth { - match tokio::time::timeout( - tokio::time::Duration::from_secs(20), - sender_reader.recv(), - ) - .await - { - Ok(Ok(msg)) => { - let reply_text = String::from_utf8_lossy(&msg.message); - info!("Received reply: {}", reply_text); - assert!(reply_text.contains("Reply")); - reply_count += 1; - } - Err(_) => { - warn!("Timeout waiting for reply {}", reply_count); - if reply_count == 0 { - panic!("No replies received"); - } - break; - } - Ok(Err(e)) => { - warn!("Error receiving reply: {}", e); - break; - } - } - } - - Ok::<_, Error>(reply_count) - }); - - let _ = sender_task.await??; - let (_, _, received_count) = receiver_task.await??; - let reply_count = reply_reader_task.await??; - - info!( - "Received {} messages, {} replies", - received_count, reply_count - ); - assert!(received_count >= 3, "Should receive at least 3 messages"); - assert!(reply_count >= 1, "Should receive at least 1 reply"); - - Ok(()) - } -} diff --git a/sdk/rust/nym-sdk/src/stream_wrapper/mixnet_stream_wrapper_ipr.rs b/sdk/rust/nym-sdk/src/stream_wrapper/mixnet_stream_wrapper_ipr.rs deleted file mode 100644 index 9807a75d39..0000000000 --- a/sdk/rust/nym-sdk/src/stream_wrapper/mixnet_stream_wrapper_ipr.rs +++ /dev/null @@ -1,937 +0,0 @@ -// Copyright 2024 - Nym Technologies SA -// SPDX-License-Identifier: GPL-3.0-only - -use super::mixnet_stream_wrapper::{MixStream, MixStreamReader, MixStreamWriter}; -use super::network_env::NetworkEnvironment; -use crate::ip_packet_client::{ - helpers::check_ipr_message_version, IprListener, MixnetMessageOutcome, -}; -use crate::{mixnet::Recipient, Error}; -use std::collections::HashMap; - -use bytes::Bytes; -use futures::StreamExt; -use nym_crypto::asymmetric::ed25519; -use nym_ip_packet_requests::{ - v8::{ - request::IpPacketRequest, - response::{ConnectResponseReply, ControlResponse, IpPacketResponse, IpPacketResponseData}, - }, - IpPair, -}; -use nym_network_defaults::ApiUrl; -use nym_validator_client::nym_api::NymApiClientExt; -use std::io; -use std::pin::Pin; -use std::task::{Context, Poll}; -use std::time::Duration; -use tokio::io::{AsyncRead, AsyncWrite, ReadBuf}; -use tokio::sync::oneshot; -use tracing::{debug, error, info}; - -const IPR_CONNECT_TIMEOUT: Duration = Duration::from_secs(60); - -#[derive(Clone)] -pub struct IprWithPerformance { - pub(crate) address: Recipient, - pub(crate) identity: ed25519::PublicKey, - pub(crate) performance: u8, -} - -#[derive(Debug, Clone, PartialEq)] -pub enum ConnectionState { - Disconnected, - Connecting, - Connected, -} - -/// Create a Nym API client with the provided URLs. -/// -/// # Arguments -/// * `nym_api_urls` - Vector of API URLs to use for the client -/// -/// # Returns -/// Configured `nym_http_api_client::Client` or an error if URLs are invalid or empty -#[allow(clippy::result_large_err)] -fn create_nym_api_client(nym_api_urls: Vec) -> Result { - let user_agent = format!("nym-sdk/{}", env!("CARGO_PKG_VERSION")); - - let urls = nym_api_urls - .into_iter() - .map(|url| url.url.parse()) - .collect::, _>>() - .map_err(|err| { - error!("malformed nym-api url: {err}"); - Error::NoNymAPIUrl - })?; - - if urls.is_empty() { - return Err(Error::NoNymAPIUrl); - } - - let client = nym_http_api_client::ClientBuilder::new_with_urls(urls)? - .with_user_agent(user_agent) - .build()?; - - Ok(client) -} - -/// Retrieve all exit nodes with their performance scores. -/// -/// Queries the network for all described nodes and filters for those with IP packet router -/// capabilities, combining node information with performance metadata. -/// -/// # Arguments -/// * `client` - Nym API client to use for queries -/// -/// # Returns -/// Vector of `IprWithPerformance` containing exit node addresses, identities, and performance scores -async fn retrieve_exit_nodes_with_performance( - client: nym_http_api_client::Client, -) -> Result, Error> { - // retrieve all nym-nodes on the network - let all_nodes = client - .get_all_described_nodes() - .await? - .into_iter() - .map(|described| (described.ed25519_identity_key(), described)) - .collect::>(); - - // annoyingly there's no convenient way of doing this in a single query - // retrieve performance scores of all exit gateways - let exit_gateways = client - .get_all_basic_nodes_with_metadata() - .await? - .nodes; - - let mut described = Vec::new(); - - for exit in exit_gateways { - if let Some(ipr_info) = all_nodes - .get(&exit.ed25519_identity_pubkey) - .and_then(|n| n.description.ip_packet_router.clone()) - { - if let Ok(parsed_address) = ipr_info.address.parse() { - described.push(IprWithPerformance { - address: parsed_address, - identity: exit.ed25519_identity_pubkey, - performance: exit.performance.round_to_integer(), - }) - } - } - } - - Ok(described) -} - -/// Select a random IPR (IP Packet Router) from available exit gateways. -/// -/// Currently selects the gateway with the highest performance score. -/// -/// # Arguments -/// * `client` - Nym API client to use for gateway discovery -/// -/// # Returns -/// `Recipient` address of the selected IPR -async fn get_random_ipr(client: nym_http_api_client::Client) -> Result { - let nodes = retrieve_exit_nodes_with_performance(client).await?; - info!("Found {} Exit Gateways", nodes.len()); - - // JS: I'm leaving the old behaviour here of choosing node with the highest performance, - // but I think you should reconsider making a pseudorandom selection weighted by some scaled performance - // otherwise all clients might end up choosing exactly the same node (I will leave this as PR comment when I get here : D) - let selected_gateway = nodes - .into_iter() - .max_by_key(|gw| gw.performance) - .ok_or_else(|| Error::NoGatewayAvailable)?; - - let ipr_address = selected_gateway.address; - - info!( - "Using IPR: {} (Gateway: {}, Performance: {:?})", - ipr_address, selected_gateway.identity, selected_gateway.performance - ); - - Ok(ipr_address) -} - -/// A bidirectional stream for sending and receiving IP packets through the mixnet. -/// -/// Manages connection to an IP Packet Router (IPR), handles tunnel establishment, -/// and maintains allocated IP addresses. Implements `AsyncRead` and `AsyncWrite` for -/// standard async I/O operations. -/// -/// # Example -/// ```no_run -/// use nym_sdk::stream_wrapper::{IpMixStream, NetworkEnvironment}; -/// -/// #[tokio::main] -/// async fn main() -> Result<(), Box> { -/// let mut stream = IpMixStream::new(NetworkEnvironment::Mainnet).await?; -/// let ip_pair = stream.connect_tunnel().await?; -/// let packet_data = vec![0u8; 100]; -/// stream.send_ip_packet(&packet_data).await?; -/// Ok(()) -/// } -/// ``` -pub struct IpMixStream { - stream: MixStream, - ipr_address: Recipient, - listener: IprListener, - allocated_ips: Option, - connection_state: ConnectionState, -} - -impl IpMixStream { - /// Create a new IP packet router stream connected to the mixnet. - /// - /// Initializes connection to mainnet by default and selects an IPR gateway. - /// Does not establish tunnel connection - call `connect_tunnel()` separately. - /// - /// # Returns - /// New `IpMixStream` instance ready to connect - pub async fn new(env: NetworkEnvironment) -> Result { - let network_defaults = env.network_defaults(); - let api_client = create_nym_api_client(network_defaults.nym_api_urls.unwrap_or_default())?; - let ipr_address = get_random_ipr(api_client).await?; - - let stream = MixStream::new(None, Some(ipr_address), Some(env.env_file_path())).await?; - - Ok(Self { - stream, - ipr_address, - listener: IprListener::new(), - allocated_ips: None, - connection_state: ConnectionState::Disconnected, - }) - } - - /// Get the Nym network address of this stream. - /// - /// # Returns - /// Reference to the stream's `Recipient` address - pub fn nym_address(&self) -> &Recipient { - self.stream.client.nym_address() - } - - /// Send an IP packet request to the connected IPR. - /// - /// # Arguments - /// * `request` - The `IpPacketRequest` to send - /// - /// # Returns - /// `Ok(())` on success, error otherwise - async fn send_ipr_request(&mut self, request: IpPacketRequest) -> Result<(), Error> { - let request_bytes = request.to_bytes()?; - self.stream.send(&request_bytes).await - } - - /// Establish tunnel connection with the IPR and allocate IP addresses. - /// - /// Sends a connect request and waits for IP allocation response. - /// Updates connection state and stores allocated IPs on success. - /// - /// # Returns - /// `IpPair` containing allocated IPv4 and IPv6 addresses - pub async fn connect_tunnel(&mut self) -> Result { - if self.connection_state != ConnectionState::Disconnected { - return Err(Error::IprStreamClientAlreadyConnectedOrConnecting); - } - - self.connection_state = ConnectionState::Connecting; - info!("Connecting to IP packet router: {}", self.ipr_address); - - match self.connect_inner().await { - Ok(ip_pair) => { - self.allocated_ips = Some(ip_pair); - self.connection_state = ConnectionState::Connected; - info!( - "Connected to IPv4: {}, IPv6: {}", - ip_pair.ipv4, ip_pair.ipv6 - ); - Ok(ip_pair) - } - Err(e) => { - self.connection_state = ConnectionState::Disconnected; - error!("Failed to connect: {:?}", e); - Err(e) - } - } - } - - /// Internal connection logic for establishing the tunnel. - /// - /// # Returns - /// `IpPair` containing allocated IP addresses - async fn connect_inner(&mut self) -> Result { - let (request, request_id) = IpPacketRequest::new_connect_request(None); - debug!("Sending connect request with ID: {}", request_id); - - self.send_ipr_request(request).await?; - self.listen_for_connect_response(request_id).await - } - - /// Listen for and process the connect response from the IPR. - /// - /// Waits up to `IPR_CONNECT_TIMEOUT` for a response matching the request ID. - /// - /// # Arguments - /// * `request_id` - ID of the connect request to match against responses - /// - /// # Returns - /// `IpPair` containing allocated IP addresses - async fn listen_for_connect_response(&mut self, request_id: u64) -> Result { - let timeout = tokio::time::sleep(IPR_CONNECT_TIMEOUT); - tokio::pin!(timeout); - - let mut framed = self.stream.framed_read(); - - loop { - tokio::select! { - _ = &mut timeout => { - return Err(Error::IPRConnectResponseTimeout); - } - frame = framed.next() => { - match frame { - None => { - return Err(Error::IPRClientStreamClosed); - } - Some(Err(e)) => { - return Err(Error::MessageRecovery(e)); - } - Some(Ok(reconstructed)) => { - if let Err(e) = check_ipr_message_version(&reconstructed) { - return Err(Error::IPRMessageVersionCheckFailed(e.to_string())); - - } - if let Ok(response) = IpPacketResponse::from_reconstructed_message(&reconstructed) { - if response.id() == Some(request_id) { - return self.handle_connect_response(response).await; - } - else { - return Err(Error::IPRNoId) - } - } - } - } - } - } - } - } - - /// Handle the connect response from the IPR. - /// - /// Extracts IP allocation from successful response or returns error on failure. - /// - /// # Arguments - /// * `response` - The `IpPacketResponse` to process - /// - /// # Returns - /// `IpPair` on successful connection, error otherwise - async fn handle_connect_response(&self, response: IpPacketResponse) -> Result { - let control_response = match response.data { - IpPacketResponseData::Control(c) => c, - other => return Err(Error::UnexpectedResponseType(other)), - }; - - match *control_response { - ControlResponse::Connect(connect_resp) => match connect_resp.reply { - ConnectResponseReply::Success(success) => Ok(success.ips), - ConnectResponseReply::Failure(reason) => Err(Error::ConnectDenied(reason)), - }, - _ => Err(Error::UnexpectedResponseType( - IpPacketResponseData::Control(control_response.clone()), - )), - } - } - - /// Send an IP packet through the tunnel. - /// - /// Requires an active tunnel connection. - /// - /// # Arguments - /// * `packet` - Raw IP packet bytes to send - /// - /// # Returns - /// `Ok(())` on success, error if not connected or send fails - pub async fn send_ip_packet(&mut self, packet: &[u8]) -> Result<(), Error> { - if self.connection_state != ConnectionState::Connected { - return Err(Error::IprStreamClientNotConnected); - } - let request = IpPacketRequest::new_data_request(packet.to_vec().into()); - self.send_ipr_request(request).await - } - - /// Handle incoming messages from the mixnet. - /// - /// Processes reconstructed messages and extracts IP packets, disconnect signals, - /// or self-ping messages. Times out after 10 seconds if no message received. - /// - /// # Returns - /// Vector of received IP packet data as `Bytes`, empty vector if no packets or on timeout - pub async fn handle_incoming(&mut self) -> Result, Error> { - let mut framed = self.stream.framed_read(); - - match tokio::time::timeout(Duration::from_secs(10), framed.next()).await { - Ok(Some(reconstructed)) => { - match self - .listener - .handle_reconstructed_message(reconstructed?) - .await - { - Ok(Some(MixnetMessageOutcome::IpPackets(packets))) => { - info!("Extracted {} IP packets", packets.len()); - Ok(packets) - } - Ok(Some(MixnetMessageOutcome::Disconnect)) => { - info!("Received disconnect"); - self.connection_state = ConnectionState::Disconnected; - self.allocated_ips = None; - Ok(Vec::new()) - } - Ok(Some(MixnetMessageOutcome::MixnetSelfPing)) => { - debug!("Received mixnet self ping"); - Ok(Vec::new()) - } - Ok(None) => Ok(Vec::new()), - Err(e) => { - error!("Failed to handle message: {}", e); - Ok(Vec::new()) - } - } - } - _ => Ok(Vec::new()), - } - } - - /// Get the allocated IP addresses for this tunnel. - /// - /// # Returns - /// `Some(&IpPair)` if IPs are allocated, `None` otherwise - pub fn allocated_ips(&self) -> Option<&IpPair> { - self.allocated_ips.as_ref() - } - - /// Check if the tunnel is currently connected. - /// - /// # Returns - /// `true` if connected, `false` otherwise - pub fn is_connected(&self) -> bool { - self.connection_state == ConnectionState::Connected - } - - /// Disconnect inner stream client from the Mixnet - note that disconnected clients cannot currently be reconnected. - pub async fn disconnect_stream(self) { - debug!("Disconnecting"); - self.stream.client.disconnect().await; - debug!("Disconnected"); - } - - /// Split the stream into separate reader and writer halves. - /// - /// Enables concurrent read and write operations similar to `TcpStream::split()`. - /// State updates (connection status, allocated IPs) are synchronized between halves - /// via oneshot channels. - /// - /// # Returns - /// Tuple of `(IpMixStreamReader, IpMixStreamWriter)` - pub fn split(self) -> (IpMixStreamReader, IpMixStreamWriter) { - debug!("Splitting IpMixStream"); - let local_addr = *self.stream.client.nym_address(); - let (stream_reader, stream_writer) = self.stream.split(); - debug!("Split IpMixStream into Reader and Writer"); - - let (state_tx, state_rx) = oneshot::channel(); - let (ips_tx, ips_rx) = oneshot::channel(); - - ( - IpMixStreamReader { - stream_reader, - listener: self.listener, - allocated_ips: self.allocated_ips, - connection_state: self.connection_state.clone(), - state_tx: Some(state_tx), - ips_tx: Some(ips_tx), - }, - IpMixStreamWriter { - stream_writer, - local_addr, - allocated_ips: self.allocated_ips, - connection_state: self.connection_state, - state_rx: Some(state_rx), - ips_rx: Some(ips_rx), - }, - ) - } -} - -impl AsyncRead for IpMixStream { - fn poll_read( - mut self: std::pin::Pin<&mut Self>, - cx: &mut std::task::Context<'_>, - buf: &mut tokio::io::ReadBuf<'_>, - ) -> std::task::Poll> { - std::pin::Pin::new(&mut self.stream).poll_read(cx, buf) - } -} - -impl AsyncWrite for IpMixStream { - fn poll_write( - mut self: std::pin::Pin<&mut Self>, - cx: &mut std::task::Context<'_>, - buf: &[u8], - ) -> std::task::Poll> { - std::pin::Pin::new(&mut self.stream).poll_write(cx, buf) - } - - fn poll_flush( - mut self: std::pin::Pin<&mut Self>, - cx: &mut std::task::Context<'_>, - ) -> std::task::Poll> { - std::pin::Pin::new(&mut self.stream).poll_flush(cx) - } - - fn poll_shutdown( - mut self: std::pin::Pin<&mut Self>, - cx: &mut std::task::Context<'_>, - ) -> std::task::Poll> { - std::pin::Pin::new(&mut self.stream).poll_shutdown(cx) - } -} - -/// Read half of a split `IpMixStream`. -/// -/// Handles incoming messages from the mixnet and processes IP packets, disconnect -/// signals, and control messages. Synchronizes connection state changes with the -/// writer half via oneshot channels. -/// -/// Created by calling `IpMixStream::split()`. Implements `AsyncRead` for standard -/// async read operations. -pub struct IpMixStreamReader { - stream_reader: MixStreamReader, - listener: IprListener, - allocated_ips: Option, - connection_state: ConnectionState, - state_tx: Option>, - ips_tx: Option>>, -} - -impl IpMixStreamReader { - /// Get the Nym network address of this reader. - /// - /// # Returns - /// The reader's `Recipient` address - pub fn nym_address(self) -> Recipient { - *self.stream_reader.local_addr() - } - - /// Handle incoming messages from the mixnet (reader half). - /// - /// Processes reconstructed messages and extracts IP packets, disconnect signals, - /// or self-ping messages. Updates connection state and notifies writer on disconnect. - /// Times out after 10 seconds if no message received. - /// - /// # Returns - /// Vector of received IP packet data as `Bytes`, empty vector if no packets or on timeout - pub async fn handle_incoming(&mut self) -> Result, Error> { - let mut framed = self.stream_reader.framed(); - - match tokio::time::timeout(Duration::from_secs(10), framed.next()).await { - Ok(Some(reconstructed)) => { - match self - .listener - .handle_reconstructed_message(reconstructed?) - .await - { - Ok(Some(MixnetMessageOutcome::IpPackets(packets))) => { - info!("Extracted {} IP packets", packets.len()); - Ok(packets) - } - Ok(Some(MixnetMessageOutcome::Disconnect)) => { - info!("Received disconnect"); - self.connection_state = ConnectionState::Disconnected; - self.allocated_ips = None; - // Send state update to writer - if let Some(tx) = self.state_tx.take() { - let _ = tx.send(ConnectionState::Disconnected); - } - if let Some(tx) = self.ips_tx.take() { - let _ = tx.send(None); - } - Ok(Vec::new()) - } - Ok(Some(MixnetMessageOutcome::MixnetSelfPing)) => { - debug!("Received mixnet self ping"); - Ok(Vec::new()) - } - Ok(None) => Ok(Vec::new()), - Err(e) => { - error!("Failed to handle message: {}", e); - Ok(Vec::new()) - } - } - } - _ => Ok(Vec::new()), - } - } - - /// Get the allocated IP addresses (reader half). - /// - /// # Returns - /// `Some(&IpPair)` if IPs are allocated, `None` otherwise - pub fn allocated_ips(&self) -> Option<&IpPair> { - self.allocated_ips.as_ref() - } - - /// Check if the tunnel is currently connected (reader half). - /// - /// # Returns - /// `true` if connected, `false` otherwise - pub fn is_connected(&self) -> bool { - self.connection_state == ConnectionState::Connected - } -} - -impl AsyncRead for IpMixStreamReader { - fn poll_read( - mut self: Pin<&mut Self>, - cx: &mut Context<'_>, - buf: &mut ReadBuf<'_>, - ) -> Poll> { - Pin::new(&mut self.stream_reader).poll_read(cx, buf) - } -} - -/// Write half of a split `IpMixStream`. -/// -/// Handles outgoing IP packets to the mixnet. Receives connection state updates -/// from the reader half via oneshot channels to maintain synchronized state. -/// -/// Created by calling `IpMixStream::split()`. Implements `AsyncWrite` for standard -/// async write operations. -pub struct IpMixStreamWriter { - stream_writer: MixStreamWriter, - local_addr: Recipient, - allocated_ips: Option, - connection_state: ConnectionState, - state_rx: Option>, - ips_rx: Option>>, -} - -impl IpMixStreamWriter { - /// Get the Nym network address of this writer. - /// - /// # Returns - /// Reference to the writer's `Recipient` address - pub fn nym_address(&self) -> &Recipient { - &self.local_addr - } - - /// Send an IP packet request to the IPR (writer half). - /// - /// # Arguments - /// * `request` - The `IpPacketRequest` to send - /// - /// # Returns - /// `Ok(())` on success, error otherwise - async fn send_ipr_request(&mut self, request: IpPacketRequest) -> Result<(), Error> { - let request_bytes = request.to_bytes()?; - self.stream_writer.write_bytes(&request_bytes).await - } - - /// Send an IP packet through the tunnel (writer half). - /// - /// Checks for state updates from reader before sending. - /// Requires an active tunnel connection. - /// - /// # Arguments - /// * `packet` - Raw IP packet bytes to send - /// - /// # Returns - /// `Ok(())` on success, error if not connected or send fails - pub async fn send_ip_packet(&mut self, packet: &[u8]) -> Result<(), Error> { - // Check for state updates from reader - if let Some(mut rx) = self.state_rx.take() { - if let Ok(new_state) = rx.try_recv() { - self.connection_state = new_state; - } else { - self.state_rx = Some(rx); - } - } - - if let Some(mut rx) = self.ips_rx.take() { - if let Ok(new_ips) = rx.try_recv() { - self.allocated_ips = new_ips; - } else { - self.ips_rx = Some(rx); - } - } - - if self.connection_state != ConnectionState::Connected { - return Err(Error::IprStreamClientNotConnected); - } - - let request = IpPacketRequest::new_data_request(packet.to_vec().into()); - self.send_ipr_request(request).await - } - - /// Get the allocated IP addresses (writer half). - /// - /// # Returns - /// `Some(&IpPair)` if IPs are allocated, `None` otherwise - pub fn allocated_ips(&self) -> Option<&IpPair> { - self.allocated_ips.as_ref() - } - - /// Check if the tunnel is connected (writer half). - /// - /// # Returns - /// `true` if connected, `false` otherwise - pub fn is_connected(&self) -> bool { - self.connection_state == ConnectionState::Connected - } -} - -impl AsyncWrite for IpMixStreamWriter { - fn poll_write( - mut self: Pin<&mut Self>, - cx: &mut Context<'_>, - buf: &[u8], - ) -> Poll> { - Pin::new(&mut self.stream_writer).poll_write(cx, buf) - } - - fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - Pin::new(&mut self.stream_writer).poll_flush(cx) - } - - fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - Pin::new(&mut self.stream_writer).poll_shutdown(cx) - } -} - -#[cfg(test)] -mod tests { - use super::*; - use crate::ip_packet_client::helpers::{ - icmp_identifier, is_icmp_echo_reply, is_icmp_v6_echo_reply, send_ping_v4, send_ping_v6, - }; - use std::net::{Ipv4Addr, Ipv6Addr}; - - #[tokio::test] - async fn connect_to_ipr() -> Result<(), Box> { - let mut stream = IpMixStream::new(NetworkEnvironment::Mainnet).await?; - let ip_pair = stream.connect_tunnel().await?; - - let ipv4: Ipv4Addr = ip_pair.ipv4; - assert!(!ipv4.is_unspecified(), "IPv4 address should not be 0.0.0.0"); - - let ipv6: Ipv6Addr = ip_pair.ipv6; - assert!(!ipv6.is_unspecified(), "IPv6 address should not be ::"); - - assert!(stream.is_connected(), "Stream should be connected"); - assert!( - stream.allocated_ips().is_some(), - "Should have allocated IPs" - ); - - stream.disconnect_stream().await; - - Ok(()) - } - - #[tokio::test] - #[ignore] - async fn dns_ping_checks() -> Result<(), Box> { - let mut stream = IpMixStream::new(NetworkEnvironment::Mainnet).await?; - let ip_pair = stream.connect_tunnel().await?; - - info!( - "Connected with IPs - IPv4: {}, IPv6: {}", - ip_pair.ipv4, ip_pair.ipv6 - ); - - let external_v4_targets = vec![ - ("Google DNS", Ipv4Addr::new(8, 8, 8, 8)), - ("Cloudflare DNS", Ipv4Addr::new(1, 1, 1, 1)), - ("Quad9 DNS", Ipv4Addr::new(9, 9, 9, 9)), - ]; - - let external_v6_targets = vec![ - ("Google DNS", "2001:4860:4860::8888".parse::()?), - ( - "Cloudflare DNS", - "2606:4700:4700::1111".parse::()?, - ), - ("Quad9 DNS", "2620:fe::fe".parse::()?), - ]; - - let identifier = icmp_identifier(); - let mut successful_v4_pings = 0; - let mut total_v4_pings = 0; - let mut successful_v6_pings = 0; - let mut total_v6_pings = 0; - - for (name, target) in &external_v4_targets { - info!("Testing IPv4 connectivity to {} ({})", name, target); - - for seq in 0..3 { - send_ping_v4(&mut stream, &ip_pair, seq, identifier, *target).await?; - total_v4_pings += 1; - } - } - - for (name, target) in &external_v6_targets { - info!("Testing IPv6 connectivity to {} ({})", name, target); - - for seq in 0..3 { - send_ping_v6(&mut stream, &ip_pair, seq, *target).await?; - total_v6_pings += 1; - } - } - - let collect_timeout = tokio::time::sleep(Duration::from_secs(10)); - tokio::pin!(collect_timeout); - - loop { - tokio::select! { - _ = &mut collect_timeout => { - info!("Finished collecting replies"); - break; - } - result = stream.handle_incoming() => { - if let Ok(packets) = result { - for packet in packets { - if let Some((reply_id, source, dest)) = is_icmp_echo_reply(&packet) { - if reply_id == identifier && dest == ip_pair.ipv4 { - successful_v4_pings += 1; - debug!("IPv4 reply from {}", source); - } - } - - if let Some((reply_id, source, dest)) = is_icmp_v6_echo_reply(&packet) { - if reply_id == identifier && dest == ip_pair.ipv6 { - successful_v6_pings += 1; - debug!("IPv6 reply from {}", source); - } - } - } - } - } - } - } - - let v4_success_rate = (successful_v4_pings as f64 / total_v4_pings as f64) * 100.0; - let v6_success_rate = (successful_v6_pings as f64 / total_v6_pings as f64) * 100.0; - - info!( - "IPv4 external connectivity: {}/{} pings successful ({:.1}%)", - successful_v4_pings, total_v4_pings, v4_success_rate - ); - info!( - "IPv6 external connectivity: {}/{} pings successful ({:.1}%)", - successful_v6_pings, total_v6_pings, v6_success_rate - ); - - assert!(successful_v4_pings > 0, "No IPv4 pings successful"); - assert!( - v4_success_rate >= 75.0, - "IPv4 success rate < 75% (got {:.1}%)", - v4_success_rate - ); - - assert!(successful_v6_pings > 0, "No IPv6 pings successful"); - assert!( - v6_success_rate >= 75.0, - "IPv6 success rate < 75% (got {:.1}%)", - v6_success_rate - ); - - stream.disconnect_stream().await; - Ok(()) - } - - #[tokio::test] - #[ignore] - async fn split_dns_ping_checks() -> Result<(), Box> { - let mut stream = IpMixStream::new(NetworkEnvironment::Mainnet).await?; - let ip_pair = stream.connect_tunnel().await?; - - info!( - "Connected with IPs - IPv4: {}, IPv6: {}", - ip_pair.ipv4, ip_pair.ipv6 - ); - - let (mut reader, mut writer) = stream.split(); - - let external_v4_targets = vec![("Google DNS", Ipv4Addr::new(8, 8, 8, 8))]; - - let identifier = icmp_identifier(); - let mut successful_v4_pings = 0; - let mut total_v4_pings = 0; - - for (name, target) in &external_v4_targets { - info!("Testing IPv4 connectivity to {} ({})", name, target); - - for seq in 0..2 { - use crate::ip_packet_client::helpers::{ - create_icmpv4_echo_request, wrap_icmp_in_ipv4, - }; - use nym_ip_packet_requests::codec::MultiIpPacketCodec; - use pnet_packet::Packet; - - let icmp_echo_request = create_icmpv4_echo_request(seq, identifier)?; - let ipv4_packet = wrap_icmp_in_ipv4(icmp_echo_request, ip_pair.ipv4, *target)?; - let bundled_packet = - MultiIpPacketCodec::bundle_one_packet(ipv4_packet.packet().to_vec().into()); - - writer.send_ip_packet(&bundled_packet).await?; - total_v4_pings += 1; - } - } - - let collect_timeout = tokio::time::sleep(Duration::from_secs(10)); - tokio::pin!(collect_timeout); - - loop { - tokio::select! { - _ = &mut collect_timeout => { - info!("Finished collecting responses"); - break; - } - result = reader.handle_incoming() => { - if let Ok(packets) = result { - for packet in packets { - if let Some((reply_id, source, dest)) = is_icmp_echo_reply(&packet) { - if reply_id == identifier && dest == ip_pair.ipv4 { - successful_v4_pings += 1; - debug!("IPv4 reply from {}", source); - } - } - } - } - } - } - } - - let v4_success_rate = if total_v4_pings > 0 { - (successful_v4_pings as f64 / total_v4_pings as f64) * 100.0 - } else { - 0.0 - }; - - info!( - "Split test - IPv4 external connectivity: {}/{} pings successful ({:.1}%)", - successful_v4_pings, total_v4_pings, v4_success_rate - ); - - assert!(successful_v4_pings > 0, "No pings successful"); - assert!( - v4_success_rate >= 75.0, - "IPv4 success rate < 75% (got {:.1}%)", - v4_success_rate - ); - - Ok(()) - } -} diff --git a/sdk/rust/nym-sdk/src/stream_wrapper/network_env.rs b/sdk/rust/nym-sdk/src/stream_wrapper/network_env.rs deleted file mode 100644 index f7a4bf3601..0000000000 --- a/sdk/rust/nym-sdk/src/stream_wrapper/network_env.rs +++ /dev/null @@ -1,55 +0,0 @@ -use std::fs; -use std::path::PathBuf; - -#[derive(Default, Debug, Clone, Copy, PartialEq, Eq)] -pub enum NetworkEnvironment { - #[default] - Mainnet, - // Sandbox, -} - -fn find_workspace_root() -> PathBuf { - let mut current = PathBuf::from(env!("CARGO_MANIFEST_DIR")); - - loop { - let cargo_toml = current.join("Cargo.toml"); - - if cargo_toml.exists() { - if let Ok(contents) = fs::read_to_string(&cargo_toml) { - // Check if this Cargo.toml defines a workspace - if contents.contains("[workspace]") { - return current; - } - } - } - - if !current.pop() { - panic!("Could not find workspace root"); - } - } -} - -impl NetworkEnvironment { - pub fn env_file_path(&self) -> PathBuf { - let root = find_workspace_root(); - match self { - Self::Mainnet => root.join("envs/mainnet.env"), - // Self::Sandbox => root.join("envs/sandbox.env"), - } - } - - pub fn network_defaults(&self) -> crate::NymNetworkDetails { - match self { - Self::Mainnet => crate::NymNetworkDetails::new_mainnet(), - // Self::Sandbox => crate::NymNetworkDetails::new_sandbox(), // TODO - } - } - - pub fn parse_network(s: &str) -> Result { - match s.to_lowercase().as_str() { - "mainnet" | "main" => Ok(Self::Mainnet), - // "sandbox" | "sand" => Ok(Self::Sandbox), - _ => Err(format!("Unknown env: {}", s)), - } - } -}