From 83dcf3fd132673f40e8766df86c32adde848d2f7 Mon Sep 17 00:00:00 2001 From: mfahampshire Date: Wed, 9 Jul 2025 15:14:16 +0200 Subject: [PATCH] got ipr wrapper connected --- sdk/rust/nym-sdk/src/stream_wrapper.rs | 2 +- .../src/stream_wrapper/connection-logics.md | 73 +++ .../stream_wrapper/mixnet_stream_wrapper.rs | 13 +- .../mixnet_stream_wrapper_ipr.rs | 421 +++++++++++++++++- 4 files changed, 489 insertions(+), 20 deletions(-) create mode 100644 sdk/rust/nym-sdk/src/stream_wrapper/connection-logics.md diff --git a/sdk/rust/nym-sdk/src/stream_wrapper.rs b/sdk/rust/nym-sdk/src/stream_wrapper.rs index c032cd6842..cfdceab5bb 100644 --- a/sdk/rust/nym-sdk/src/stream_wrapper.rs +++ b/sdk/rust/nym-sdk/src/stream_wrapper.rs @@ -4,4 +4,4 @@ mod mixnet_stream_wrapper; mod mixnet_stream_wrapper_ipr; pub use mixnet_stream_wrapper::{MixSocket, MixStream, MixStreamReader, MixStreamWriter}; -// pub use mixnet_stream_wrapper_ipr::{} TODO +pub use mixnet_stream_wrapper_ipr::{IpMixSocket, IpMixStream}; diff --git a/sdk/rust/nym-sdk/src/stream_wrapper/connection-logics.md b/sdk/rust/nym-sdk/src/stream_wrapper/connection-logics.md new file mode 100644 index 0000000000..849b04d47d --- /dev/null +++ b/sdk/rust/nym-sdk/src/stream_wrapper/connection-logics.md @@ -0,0 +1,73 @@ + +## MixStream +WRITING (App → Mixnet): +1. App calls stream.write(bytes) +2. AsyncWrite implementation: + - Deserializes bytes as InputMessage using InputMessageCodec + - Forwards InputMessage to MixnetClient +3. MixnetClient sends to mixnet + +READING (Mixnet → App): +1. MixnetClient receives ReconstructedMessage from Mixnet +2. Stream implementation buffers messages +3. AsyncRead implementation: + - Encodes ReconstructedMessage using ReconstructedMessageCodec + - Returns encoded bytes to app + +Flow: App bytes ↔ InputMessage/ReconstructedMessage ↔ Mixnet + +## IP Client +CONNECTING: +1. Create MixnetClient +2. Send IpPacketRequest::Connect wrapped in InputMessage +3. Use wait_for_messages() to get ReconstructedMessage from Mixnet +4. Parse as IpPacketResponse::Connect +5. Extract allocated IPs + +SENDING IP PACKETS: +1. App provides IP packet bytes +2. Wrap in IpPacketRequest::Data +3. Send as InputMessage through MixnetClient + +RECEIVING IP PACKETS: +1. IprListener processes ReconstructedMessages +2. Parse as IpPacketResponse +3. Extract IP packets from IpPacketResponse::Data using MultiIpPacketCodec +4. Return raw IP packet bytes to app + +Flow: IP packets → IpPacketRequest → InputMessage → Mixnet → IPR + IPR → Mixnet → IpPacketResponse → IP packets + +## IPRMixStream +SETUP: +1. Create MixStream to IPR address +2. Split into reader/writer +3. Background task owns reader, processes IpPacketResponse messages +4. Main struct owns writer for sending + +CONNECTING +1. Send IpPacketRequest::Connect wrapped in InputMessage +2. Use wait_for_messages() to get ReconstructedMessage from Mixnet +3. Parse as IpPacketResponse::Connect +4. Extract allocated IPs + +WRITING (TUN → IPR?): +1. TUN device calls stream.write(ip_packet_bytes) +2. AsyncWrite implementation: + - Wrap IP packet in IpPacketRequest::Data + - Serialize to bytes using to_bytes() + - Call writer.write_bytes() which: + - Wraps in InputMessage + - Sends through mixnet + +READING (IPR → TUN): +1. Background task continuously reads from MixStreamReader +2. Decodes bytes as ReconstructedMessage +3. Parses as IpPacketResponse +4. Extracts IP packets from Data responses +5. Adds to shared buffer +6. AsyncRead polls buffer and returns IP packets to TUN + +Flow: +Write: TUN IP packet → IpPacketRequest → serialize → MixStreamWriter → InputMessage → Mixnet → IPR +Read: IPR → Mixnet → MixStreamReader → ReconstructedMessage → IpPacketResponse → extract IP packets → buffer → TUN 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 index 715548cf8b..2b2834b2d6 100644 --- a/sdk/rust/nym-sdk/src/stream_wrapper/mixnet_stream_wrapper.rs +++ b/sdk/rust/nym-sdk/src/stream_wrapper/mixnet_stream_wrapper.rs @@ -27,14 +27,16 @@ use tracing::{debug, info, warn}; /// However, we can't map this one to one onto the TcpSocket as there isn't really a concept of binding to a port with the MixnetClient; it connects to its Gateway and then just accepts incoming messages from the Gw via the Websocket connection. However, we can stick with the idea of creating a Socket in an unconnected state, either using it to make a new Stream (connecting it to its EntryGw) or connecting it *to* something (once the IPR functionality is enabled, this will mean the creation of a Stream + kicking off the creation of a tunnel to an ExitGw + IPR). /// The cause for a MixSocket > going striaght to a MixStream is creating a Nym Client disconnected from the Mixnet first, then upgrading to a Stream when connecting it. Once LP is implemented, this will also allow us to follow something like what is implemented for the Tokio::net::UdpFramed abstraction, where we can create multiple MixStream instances from a single MixSocket, all connected to different Recipients. pub struct MixSocket { - inner: MixnetClient, + pub inner: MixnetClient, } impl MixSocket { /// Create a new socket that is disconnected from the Mixnet - kick off the Mixnet client with config for builder. /// Following idea of having single client with multiple concurrent connections represented by per-Recipient MixStream instance. pub async fn new() -> Result { - todo!() + // TODO make this take an option, if Some create with builder pattern, if None make ephemeral like we already are. + let inner = MixnetClient::connect_new().await?; + Ok(MixSocket { inner }) } /// Connect to a specific peer (Nym Client) and return a Stream (cf TcpSocket::connect() / TcpStream::new()). @@ -61,7 +63,7 @@ impl MixSocket { } pub struct MixStream { - client: MixnetClient, + pub client: MixnetClient, peer: Option, // We might be accepting incoming messages and replying, so might not have a Nym addr to talk to.. peer_surbs: Option, // ..since we might just be using SURBs instead } @@ -158,6 +160,11 @@ impl MixStream { Ok(()) } + /// Sidesteps the AsyncRead/codec in place of using the Mixnet Client's message-based functionality for getting messages: using to debug connection with the IPR Stream Wrapper. + pub async fn wait_for_messages(&mut self) -> Option> { + self.client.wait_for_messages().await + } + /// Disconnect client from the Mixnet - note that disconnected clients cannot currently be reconnected. pub async fn disconnect(self) { debug!("Disconnecting"); 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 index cc63e3b10a..6dfea874f6 100644 --- 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 @@ -1,22 +1,411 @@ -use super::mixnet_stream_wrapper::{MixSocket, MixStream, MixStreamReader, MixStreamWriter}; -use futures::StreamExt; -use nym_gateway_directory::IpPacketRouterAddress; +use crate::UserAgent; +use crate::{mixnet::Recipient, Error}; + +use super::mixnet_stream_wrapper::{MixSocket, MixStream}; + +use bytes::Bytes; +use bytes::BytesMut; +use nym_gateway_directory::{ + Config as GatewayConfig, GatewayClient, GatewayType, IpPacketRouterAddress, +}; use nym_ip_packet_requests::{ - codec::MultiIpPacketCodec, v8::{ - request::{ControlRequest, IpPacketRequest, IpPacketRequestData}, - response::{ - ConnectResponse, ConnectResponseReply, ControlResponse, InfoLevel, IpPacketResponse, - IpPacketResponseData, - }, + request::IpPacketRequest, + response::{ConnectResponseReply, ControlResponse, IpPacketResponse, IpPacketResponseData}, }, IpPair, }; -use nym_sphinx::receiver::ReconstructedMessage; -use std::io; -use std::pin::Pin; -use std::task::{Context, Poll}; +use nym_sphinx::receiver::ReconstructedMessageCodec; +use std::sync::Once; use std::time::Duration; -use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, ReadBuf}; -use tokio_util::codec::FramedRead; -use tracing::{debug, error, info, warn}; +use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; +use tokio_util::codec::{Decoder, FramedRead}; +use tracing::{debug, error, info}; + +use crate::ip_packet_client::{ + helpers::check_ipr_message_version, + listener::{IprListener, MixnetMessageOutcome}, +}; + +const IPR_CONNECT_TIMEOUT: Duration = Duration::from_secs(60); + +#[derive(Debug, Clone, PartialEq)] +enum ConnectionState { + Disconnected, + Connecting, + Connected, +} + +pub struct IpMixSocket { + inner: MixSocket, + gateway_client: GatewayClient, +} + +impl IpMixSocket { + pub async fn new() -> Result { + let inner = MixSocket::new().await?; + let gateway_client = Self::create_gateway_client()?; + Ok(Self { + inner, + gateway_client, + }) + } + + fn create_gateway_client() -> Result { + let user_agent = UserAgent { + application: "nym-ipr-streamer".to_string(), + version: "0.0.1".to_string(), + platform: "xxxxxxx".to_string(), + git_commit: "".to_string(), + }; + + let mainnet_network_defaults = crate::NymNetworkDetails::default(); + let api_url = mainnet_network_defaults + .endpoints + .first() + .unwrap() + .api_url() + .unwrap(); + + let nyxd_url = mainnet_network_defaults + .endpoints + .first() + .unwrap() + .nyxd_url(); + + let nym_vpn_api_url = mainnet_network_defaults.nym_vpn_api_url().unwrap(); + + let config = GatewayConfig { + nyxd_url, + api_url, + nym_vpn_api_url: Some(nym_vpn_api_url), + min_gateway_performance: None, + mix_score_thresholds: None, + wg_score_thresholds: None, + }; + + Ok(GatewayClient::new(config, user_agent).unwrap()) + } + + async fn get_best_ipr_address(&self) -> Result { + let exit_gateways = self + .gateway_client + .lookup_gateways(GatewayType::MixnetExit) + .await + .unwrap(); + + info!("Found {} Exit Gateways", exit_gateways.len()); + + let selected_gateway = exit_gateways + .into_iter() + .filter(|gw| gw.has_ipr_address()) + .max_by_key(|gw| { + gw.mixnet_performance + .map(|p| p.round_to_integer()) + .unwrap_or(0) + }) + .unwrap(); + + let ipr_address = selected_gateway.ipr_address.unwrap(); + + info!( + "Using IPR: {} (Gateway: {}, Performance: {:?})", + ipr_address, + selected_gateway.identity(), + selected_gateway.mixnet_performance + ); + + Ok(ipr_address) + } + + pub async fn connect(&self) -> Result { + let ipr_address = self.get_best_ipr_address().await?; + let stream = MixStream::new(None, Recipient::from(ipr_address.clone())).await; + Ok(IpMixStream::new(stream, ipr_address)) + } + + pub fn nym_address(&self) -> &Recipient { + self.inner.nym_address() + } +} + +pub struct IpMixStream { + stream: MixStream, + ipr_address: IpPacketRouterAddress, + listener: IprListener, + allocated_ips: Option, + connection_state: ConnectionState, +} + +impl IpMixStream { + fn new(stream: MixStream, ipr_address: IpPacketRouterAddress) -> Self { + Self { + stream, + ipr_address, + listener: IprListener::new(), + allocated_ips: None, + connection_state: ConnectionState::Disconnected, + } + } + + async fn send_ipr_request(&mut self, request: IpPacketRequest) -> Result<(), Error> { + let request_bytes = request.to_bytes()?; + self.stream.write_bytes(&request_bytes).await + } + + pub async fn connect_tunnel(&mut self) -> Result { + if self.connection_state != ConnectionState::Disconnected { + return Err(Error::new_unsupported( + "Already connected or connecting".to_string(), + )); + } + + 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.clone()); + 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) + } + } + } + + 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 + } + + 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 buffer = vec![0u8; 65536]; + + loop { + tokio::select! { + _ = &mut timeout => { + return Err(Error::new_unsupported("Timeout waiting for connect response".to_string())); + } + result = self.stream.read(&mut buffer) => { + match result { + Ok(0) => { + debug!("Stream closed"); + return Err(Error::new_unsupported("Stream closed".to_string())); + } + Ok(n) => { + debug!("Received {} bytes", n); + + let mut codec = ReconstructedMessageCodec {}; + let mut buf = BytesMut::from(&buffer[..n]); + + if let Ok(Some(reconstructed)) = codec.decode(&mut buf) { + if let Err(e) = check_ipr_message_version(&reconstructed) { + debug!("Version check failed: {}", e); + continue; + } + if let Ok(response) = IpPacketResponse::from_reconstructed_message(&reconstructed) { + if response.id() == Some(request_id) { + return self.handle_connect_response(response).await; + } + } + } + } + Err(e) => { + error!("Read error: {}", e); + return Err(Error::new_unsupported(format!("Read error: {}", e))); + } + } + } + } + } + } + + async fn handle_connect_response(&self, response: IpPacketResponse) -> Result { + let control_response = match response.data { + IpPacketResponseData::Control(control) => control, + _ => { + return Err(Error::new_unsupported( + "Expected control response".to_string(), + )) + } + }; + + match *control_response { + ControlResponse::Connect(connect_resp) => match connect_resp.reply { + ConnectResponseReply::Success(success) => Ok(success.ips), + ConnectResponseReply::Failure(reason) => Err(Error::new_unsupported(format!( + "Connect denied: {:?}", + reason + ))), + }, + _ => Err(Error::new_unsupported( + "Unexpected control response type".to_string(), + )), + } + } + + pub async fn send_ip_packet(&mut self, packet: &[u8]) -> Result<(), Error> { + if self.connection_state != ConnectionState::Connected { + return Err(Error::new_unsupported("Not connected".to_string())); + } + + let request = IpPacketRequest::new_data_request(packet.to_vec().into()); + self.send_ipr_request(request).await + } + + pub async fn process_incoming(&mut self) -> Result, Error> { + // TODO switch to framedreading? + let mut buffer = vec![0u8; 65536]; + + match tokio::time::timeout(Duration::from_secs(10), self.stream.read(&mut buffer)).await { + Ok(Ok(n)) if n > 0 => { + debug!("Read {} bytes", n); + + let mut codec = ReconstructedMessageCodec {}; + let mut buf = BytesMut::from(&buffer[..n]); + + if let Ok(Some(reconstructed)) = codec.decode(&mut buf) { + match self + .listener + .handle_reconstructed_message(reconstructed) + .await + { + Ok(Some(MixnetMessageOutcome::IpPackets(packets))) => { + debug!("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) => { + debug!("Failed to handle message: {}", e); + Ok(Vec::new()) + } + } + } else { + Ok(Vec::new()) + } + } + _ => Ok(Vec::new()), + } + } + + pub fn allocated_ips(&self) -> Option<&IpPair> { + self.allocated_ips.as_ref() + } + + pub fn is_connected(&self) -> bool { + self.connection_state == ConnectionState::Connected + } +} + +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) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::net::{Ipv4Addr, Ipv6Addr}; + + static INIT: Once = Once::new(); + + fn init_logging() { + INIT.call_once(|| { + nym_bin_common::logging::setup_tracing_logger(); + }); + } + + #[tokio::test] + async fn connect_to_ipr() -> Result<(), Box> { + init_logging(); + + let socket = IpMixSocket::new().await?; + + let mut stream = socket.connect().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" + ); + + Ok(()) + } + + #[tokio::test] + async fn send_ping() -> Result<(), Box> { + // TODO pull in https://github.com/nymtech/nym-vpn-client/blob/develop/nym-vpn-core/crates/nym-connection-monitor/src/packet_helpers.rs#L7-L42 + // TODO pull in https://github.com/nymtech/nym-vpn-client/blob/develop/nym-vpn-core/crates/nym-gateway-probe/src/icmp.rs#L25 + init_logging(); + + let socket = IpMixSocket::new().await?; + + let mut stream = socket.connect().await?; + let ip_pair = stream.connect_tunnel().await?; + + // send ping + // + + Ok(()) + } +}