From c07ef0253d9fd858a96179f031c100d773f8b61e Mon Sep 17 00:00:00 2001 From: mfahampshire Date: Fri, 27 Mar 2026 20:35:26 +0000 Subject: [PATCH] Max/sdk stream wrapper (#6320) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * Replace MixnetStream with LP framing - Replace custom header with LpFrameHeader - Added sequence number for message ordering * IPR: support LP Stream-framed client connections - Detect and route LP Stream frames in mixnet_listener - Wrap inline responses in LP Stream frames - Thread stream_id to ConnectedClientHandler for TUN responses * sdk: add ipr_wrapper module with IpMixStream - IpMixStream wraps MixnetStream for IPR tunnel over mixnet - LP Stream framing handled automatically by MixnetStream - Gateway discovery, connect handshake, IP packet send/receive * sdk: remove superseded stream_wrapper module * Trim obvious comments, add architecture.md stub * sdk: add missing deps and fix warnings * Cut down architecture diagram until finished with rest of the code, leaving stubs * sdk: refactor IpMixStream, extract shared helpers - Extract gateway discovery and connect response parsing - Add recv() to MixnetStream, remove 64KB read buffer - Simplify IpMixStream constructor * Fix SphinxStream renames missed during rebase * Add IpPacketResponse::from_bytes() for stream-based deserialization * Clean up ip_packet_client: delete stale connect.rs, take raw bytes not ReconstructedMessage * Clippy * Delete unused ip_packet_client modules - Remove helpers.rs (ICMP utilities moved to example) - Remove error.rs (errors consolidated into sdk/error.rs) - Remove README.md - Update module root to only export discovery + listener * Simplify listener, IpMixStream, and network_env - Collapse IprListener struct into standalone handle_ipr_response() - Move check_ipr_message_version() into listener.rs - Remove IpMixStream test module (moved to example) - Remove parse_network() and commented-out Sandbox arms - Return Result from find_workspace_root() instead of panicking - Add IprTunnelDisconnected and WorkspaceRootNotFound error variants * Refactor IPR stream handling and document seq conventions - Inline stream_id tracking (remove current_stream_id field) - Re-export encode_stream_frame from clients module - Document seq=0 reservation for inline control responses - Document data-path counter starting at 1 with skip-on-wrap * Add ipr_tunnel example for integration testing - ICMP ping through IPR with --gateway flag for targeting specific exits - Move pnet_packet from dependencies to dev-dependencies * Add message reordering to stream router - Buffer out-of-order messages per-stream using BTreeMap - Drain contiguous sequences individually to preserve message boundaries - Drop duplicate/old sequence numbers with a warning - Remove dead_code allow on StreamFrame::sequence_num * Clean up comments and fill architecture.md - Remove separator line comments - Update stale comments about ordering not being implemented - Remove collapsible_if allows, use let-else instead - Fill in architecture.md data flow and connection lifecycle * Simplify ipr_tunnel example to minimal smoke test - Single ping instead of multi-ping loop - Remove identifier and PING_COUNT - Collapse ICMP helpers into single build_icmp_ping function * Add dual-stack IPv6 ping and rename gateway → ipr - Rename --gateway flag to --ipr and new_with_gateway() to new_with_ipr() - Add ICMPv6 ping to ipr_tunnel example for dual-stack smoke test - Tighten echo reply validation (protocol field check, diagnostic output) - Document IP allocation (subnets, static vs dynamic, client keying) in architecture.md - Promote LP Stream Open handshake log to INFO * Tweak subnet comment in docs * Don't stop IPR listener on decode failure - Change break to continue so garbage packets can't kill the listener - Remaining valid packets in the bundle are still processed * Fix license headers and use workspace dep for pnet_packet - Switch GPL-3.0 to Apache-2.0 on all SDK library files - Add missing license headers to 7 files - Use workspace version for pnet_packet dependency * Document IP pool isolation from WG/LP dVPN pool - IPR uses 10.0.0.0/16 on nymtun, WG uses 10.1.0.0/16 on nymwg - Reference constants.rs as source of truth * Remove network_env.rs and simplify IpMixStream API - Default to mainnet via setup_env(None) instead of requiring env param - Remove NetworkEnvironment enum and workspace root detection - Remove WorkspaceRootNotFound error variant - Update ipr_tunnel example to match new signatures * Use weighted random selection for IPR gateway discovery - Replace max_by_key with choose_weighted biased by performance score - Prevents all clients converging on a single highest-performing IPR * Cap stream reorder buffer to prevent unbounded memory growth - Add MAX_REORDER_BUFFER (256) to limit per-stream pending messages: - buffer overflows = skip ahead to lowest buffered seq and drain - protects against malicious senders that deliberately skip sequence numbers * Extract shared IPR response helpers into nym-ip-packet-requests - Add response_helpers module with version check, connect response parsing, and control response dispatch - SDK ip_packet_client now delegates to shared module - Monorepo nym-ip-packet-client uses shared version check and connect response parsing - Fix doc comment attributing fork to nym-vpn-client * Extract ICMP test helpers into nym-ip-packet-requests - Add icmp_utils module behind test-utils feature flag - Move build_icmp_ping, build_icmpv6_ping, is_echo_reply_v4/v6 from example - Update ipr_tunnel example to use shared helpers * Add protocol v9 LP-framed transport marker - Add v9 module (re-exports v8, VERSION=9) - Accept v9 requests and responses in IPR - Switch SDK IpMixStream to send v9 * Log protocol version in dynamic connect requests * Remove KCP from IPR and fix unwrap_or_default in SDK - Remove all KCP session management from ip-packet-router (replaced by LP Stream framing) - Drop nym-kcp dependency and KcpError variant from IPR - Replace unwrap_or_default with ok_or(Error::NoNymAPIUrl) in IpMixStream::new() * Add v9 protocol wrapper constructors and enforce version/transport consistency - Add v9::new_connect_request(), new_data_request(), new_ip_packet_response() to centralise version stamping - Replace manual protocol.version overrides in SDK and IPR with v9 wrapper calls - Bump nym-ip-packet-client current re-export from v8 to v9 - Enforce LP Stream frames must carry v9+ payloads, non-stream must be v8 or lower * Filter IPR exit nodes by minimum v9-compatible release version - Define MIN_RELEASE_VERSION (1.30.0) in ip-packet-requests/v9 alongside protocol constants - Add semver-based filtering in SDK gateway discovery to skip nodes below v9 threshold - Add semver dependency to ip-packet-requests and nym-sdk * Use numeric version comparison for transport/version enforcement - Compare version as u8 instead of enum equality so future v10+ is handled correctly - Remove unused `use super::*` import left over from KCP test removal --- Cargo.lock | 8 +- common/ip-packet-requests/Cargo.toml | 6 + common/ip-packet-requests/src/icmp_utils.rs | 117 ++++ common/ip-packet-requests/src/lib.rs | 6 + .../src/response_helpers.rs | 134 +++++ common/ip-packet-requests/src/v8/response.rs | 8 +- common/ip-packet-requests/src/v9/mod.rs | 34 ++ nym-ip-packet-client/src/connect.rs | 45 +- nym-ip-packet-client/src/helpers.rs | 36 +- nym-ip-packet-client/src/lib.rs | 2 +- sdk/rust/nym-sdk/Cargo.toml | 5 + sdk/rust/nym-sdk/examples/ipr_tunnel.rs | 92 ++++ sdk/rust/nym-sdk/src/error.rs | 36 ++ sdk/rust/nym-sdk/src/ip_packet_client.rs | 17 + .../nym-sdk/src/ip_packet_client/discovery.rs | 121 +++++ .../nym-sdk/src/ip_packet_client/listener.rs | 20 + sdk/rust/nym-sdk/src/ipr_wrapper.rs | 13 + .../nym-sdk/src/ipr_wrapper/architecture.md | 80 +++ .../nym-sdk/src/ipr_wrapper/ip_mix_stream.rs | 192 +++++++ sdk/rust/nym-sdk/src/lib.rs | 6 + .../src/mixnet/stream/mixnet_stream.rs | 14 + sdk/rust/nym-sdk/src/mixnet/stream/mod.rs | 113 +++- .../nym-sdk/src/mixnet/stream/protocol.rs | 4 +- service-providers/ip-packet-router/Cargo.toml | 2 +- .../src/clients/connected_client_handler.rs | 61 ++- .../ip-packet-router/src/clients/mod.rs | 2 +- .../ip-packet-router/src/error.rs | 4 +- .../ip-packet-router/src/ip_packet_router.rs | 1 - .../src/kcp_session_manager.rs | 504 ------------------ service-providers/ip-packet-router/src/lib.rs | 1 - .../ip-packet-router/src/messages/mod.rs | 2 + .../ip-packet-router/src/messages/request.rs | 26 + .../src/messages/request/v9.rs | 27 + .../ip-packet-router/src/messages/response.rs | 1 + .../ip-packet-router/src/mixnet_listener.rs | 471 +++++++--------- 35 files changed, 1338 insertions(+), 873 deletions(-) create mode 100644 common/ip-packet-requests/src/icmp_utils.rs create mode 100644 common/ip-packet-requests/src/response_helpers.rs create mode 100644 common/ip-packet-requests/src/v9/mod.rs create mode 100644 sdk/rust/nym-sdk/examples/ipr_tunnel.rs create mode 100644 sdk/rust/nym-sdk/src/ip_packet_client.rs create mode 100644 sdk/rust/nym-sdk/src/ip_packet_client/discovery.rs create mode 100644 sdk/rust/nym-sdk/src/ip_packet_client/listener.rs create mode 100644 sdk/rust/nym-sdk/src/ipr_wrapper.rs create mode 100644 sdk/rust/nym-sdk/src/ipr_wrapper/architecture.md create mode 100644 sdk/rust/nym-sdk/src/ipr_wrapper/ip_mix_stream.rs delete mode 100644 service-providers/ip-packet-router/src/kcp_session_manager.rs create mode 100644 service-providers/ip-packet-router/src/messages/request/v9.rs diff --git a/Cargo.lock b/Cargo.lock index f24cf1b3a7..59c16f3898 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6957,12 +6957,15 @@ dependencies = [ "nym-crypto", "nym-service-provider-requests-common", "nym-sphinx", + "pnet_packet", "rand 0.8.5", + "semver 1.0.27", "serde", "thiserror 2.0.12", "time", "tokio", "tokio-util", + "tracing", ] [[package]] @@ -6985,7 +6988,7 @@ dependencies = [ "nym-exit-policy", "nym-id", "nym-ip-packet-requests", - "nym-kcp", + "nym-lp", "nym-network-defaults", "nym-network-requester", "nym-sdk", @@ -7792,6 +7795,7 @@ dependencies = [ "nym-crypto", "nym-gateway-requests", "nym-http-api-client", + "nym-ip-packet-requests", "nym-lp", "nym-network-defaults", "nym-ordered-buffer", @@ -7805,8 +7809,10 @@ dependencies = [ "nym-topology", "nym-validator-client", "parking_lot", + "pnet_packet", "rand 0.8.5", "reqwest 0.13.1", + "semver 1.0.27", "serde", "tap", "tempfile", diff --git a/common/ip-packet-requests/Cargo.toml b/common/ip-packet-requests/Cargo.toml index 487ff4d782..0d062b2d45 100644 --- a/common/ip-packet-requests/Cargo.toml +++ b/common/ip-packet-requests/Cargo.toml @@ -10,7 +10,11 @@ license.workspace = true description = "Codec, signing functionality, and different version definitions for IP packet request and responses" +[features] +test-utils = ["pnet_packet"] + [dependencies] +pnet_packet = { workspace = true, optional = true } bincode = { workspace = true } bytes = { workspace = true } nym-bin-common = { workspace = true } @@ -18,8 +22,10 @@ nym-crypto = { workspace = true } nym-service-provider-requests-common = { workspace = true } nym-sphinx = { workspace = true } rand = { workspace = true } +semver = { workspace = true } serde = { workspace = true, features = ["derive"] } thiserror = { workspace = true } time = { workspace = true } tokio = { workspace = true, features = ["time"] } tokio-util = { workspace = true, features = ["codec"] } +tracing = { workspace = true } diff --git a/common/ip-packet-requests/src/icmp_utils.rs b/common/ip-packet-requests/src/icmp_utils.rs new file mode 100644 index 0000000000..9dd9b1baea --- /dev/null +++ b/common/ip-packet-requests/src/icmp_utils.rs @@ -0,0 +1,117 @@ +// Copyright 2025 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +// Extracted from sdk/rust/nym-sdk/examples/ipr_tunnel.rs + +//! ICMP/ICMPv6 packet construction and reply detection helpers for testing +//! IPR connectivity. Gated behind the `test-utils` feature. + +use std::net::{Ipv4Addr, Ipv6Addr}; + +use pnet_packet::Packet; +use pnet_packet::icmp::echo_reply::EchoReplyPacket; +use pnet_packet::icmp::echo_request::MutableEchoRequestPacket; +use pnet_packet::icmp::{IcmpPacket, IcmpTypes}; +use pnet_packet::icmpv6::Icmpv6Types; +use pnet_packet::ipv4::{Ipv4Flags, MutableIpv4Packet}; +use pnet_packet::ipv6::MutableIpv6Packet; + +/// Build a complete IPv4 ICMP echo request packet. +pub fn build_icmp_ping(src: Ipv4Addr, dst: Ipv4Addr, seq: u16) -> Option> { + let mut echo = MutableEchoRequestPacket::owned(vec![0u8; 64])?; + echo.set_icmp_type(IcmpTypes::EchoRequest); + echo.set_icmp_code(pnet_packet::icmp::IcmpCode::new(0)); + echo.set_sequence_number(seq); + let cksum = pnet_packet::icmp::checksum(&IcmpPacket::new(echo.packet())?); + echo.set_checksum(cksum); + + let total_len = 20 + echo.packet().len(); + let mut ip = MutableIpv4Packet::owned(vec![0u8; total_len])?; + ip.set_version(4); + ip.set_header_length(5); + ip.set_total_length(total_len as u16); + ip.set_ttl(64); + ip.set_next_level_protocol(pnet_packet::ip::IpNextHeaderProtocols::Icmp); + ip.set_source(src); + ip.set_destination(dst); + ip.set_flags(Ipv4Flags::DontFragment); + ip.set_payload(echo.packet()); + + let mut buf = ip.consume_to_immutable().packet().to_vec(); + let cksum = ipv4_checksum(&buf); + buf[10] = (cksum >> 8) as u8; + buf[11] = cksum as u8; + Some(buf) +} + +/// Build a complete IPv6 ICMPv6 echo request packet. +pub fn build_icmpv6_ping(src: Ipv6Addr, dst: Ipv6Addr, seq: u16) -> Option> { + let mut echo = + pnet_packet::icmpv6::echo_request::MutableEchoRequestPacket::owned(vec![0u8; 64])?; + echo.set_icmpv6_type(Icmpv6Types::EchoRequest); + echo.set_icmpv6_code(pnet_packet::icmpv6::Icmpv6Code::new(0)); + echo.set_sequence_number(seq); + let cksum = pnet_packet::icmpv6::checksum( + &pnet_packet::icmpv6::Icmpv6Packet::new(echo.packet())?, + &src, + &dst, + ); + echo.set_checksum(cksum); + + let payload_len = echo.packet().len(); + let mut ip = MutableIpv6Packet::owned(vec![0u8; 40 + payload_len])?; + ip.set_version(6); + ip.set_payload_length(payload_len as u16); + ip.set_next_header(pnet_packet::ip::IpNextHeaderProtocols::Icmpv6); + ip.set_hop_limit(64); + ip.set_source(src); + ip.set_destination(dst); + ip.set_payload(echo.packet()); + + Some(ip.consume_to_immutable().packet().to_vec()) +} + +/// Check if a raw packet is an IPv4 ICMP echo reply destined to `expected_dst`. +pub fn is_echo_reply_v4(data: &[u8], expected_dst: Ipv4Addr) -> bool { + let Some(ip) = pnet_packet::ipv4::Ipv4Packet::new(data) else { + return false; + }; + if ip.get_destination() != expected_dst { + return false; + } + if ip.get_next_level_protocol() != pnet_packet::ip::IpNextHeaderProtocols::Icmp { + return false; + } + let Some(reply) = EchoReplyPacket::new(ip.payload()) else { + return false; + }; + reply.get_icmp_type() == IcmpTypes::EchoReply +} + +/// Check if a raw packet is an IPv6 ICMPv6 echo reply destined to `expected_dst`. +pub fn is_echo_reply_v6(data: &[u8], expected_dst: Ipv6Addr) -> bool { + let Some(ip) = pnet_packet::ipv6::Ipv6Packet::new(data) else { + return false; + }; + if ip.get_destination() != expected_dst { + return false; + } + if ip.get_next_header() != pnet_packet::ip::IpNextHeaderProtocols::Icmpv6 { + return false; + } + let Some(reply) = pnet_packet::icmpv6::echo_reply::EchoReplyPacket::new(ip.payload()) else { + return false; + }; + reply.get_icmpv6_type() == Icmpv6Types::EchoReply +} + +fn ipv4_checksum(header: &[u8]) -> u16 { + let mut sum = 0u32; + for i in (0..20).step_by(2) { + sum += ((header[i] as u32) << 8) | header[i + 1] as u32; + } + while (sum >> 16) > 0 { + sum = (sum & 0xFFFF) + (sum >> 16); + } + !sum as u16 +} diff --git a/common/ip-packet-requests/src/lib.rs b/common/ip-packet-requests/src/lib.rs index 977a277446..f3014a69ec 100644 --- a/common/ip-packet-requests/src/lib.rs +++ b/common/ip-packet-requests/src/lib.rs @@ -3,10 +3,14 @@ use std::fmt::{Display, Formatter}; use std::net::{Ipv4Addr, Ipv6Addr}; pub mod codec; +#[cfg(feature = "test-utils")] +pub mod icmp_utils; +pub mod response_helpers; pub mod sign; pub mod v6; pub mod v7; pub mod v8; +pub mod v9; // version 3: initial version // version 4: IPv6 support @@ -14,6 +18,8 @@ pub mod v8; // version 6: Increase the available IPs // version 7: Add signature support (for the future) // version 8: Anonymous sends +// version 9: LP-framed transport (SphinxStream) +// response_helpers: shared IPR response parsing (nym-ip-packet-client + nym-sdk) #[derive(Copy, Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)] pub struct IpPair { diff --git a/common/ip-packet-requests/src/response_helpers.rs b/common/ip-packet-requests/src/response_helpers.rs new file mode 100644 index 0000000000..5caa81ab0d --- /dev/null +++ b/common/ip-packet-requests/src/response_helpers.rs @@ -0,0 +1,134 @@ +// Copyright 2025 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use bytes::{Bytes, BytesMut}; +use tokio_util::codec::Decoder; +use tracing::{error, info, warn}; + +use crate::{ + IpPair, + codec::MultiIpPacketCodec, + v8::response::{ + ConnectResponseReply, ControlResponse, InfoLevel, IpPacketResponse, IpPacketResponseData, + }, +}; + +#[derive(Debug, thiserror::Error)] +pub enum IprResponseError { + #[error("no version byte in message")] + NoVersionByte, + + #[error("version mismatch: received v{received}, expected v{expected}")] + VersionMismatch { expected: u8, received: u8 }, + + #[error("expected control response, got {0:?}")] + UnexpectedResponse(IpPacketResponseData), + + #[error("connect denied: {0:?}")] + ConnectDenied(crate::v8::response::ConnectFailureReason), +} + +pub enum MixnetMessageOutcome { + IpPackets(Vec), + Disconnect, +} + +// Extracted from: +// nym-ip-packet-client/src/helpers.rs — check_ipr_message_version() +// sdk/rust/nym-sdk/src/ip_packet_client/listener.rs — check_ipr_message_version() +/// Check that the first byte of an IPR message matches the expected protocol version. +pub fn check_ipr_message_version(data: &[u8], expected: u8) -> Result<(), IprResponseError> { + let version = data.first().ok_or(IprResponseError::NoVersionByte)?; + if *version != expected { + return Err(IprResponseError::VersionMismatch { + expected, + received: *version, + }); + } + Ok(()) +} + +// Extracted from: +// nym-ip-packet-client/src/connect.rs — handle_connect_response() + handle_ip_packet_router_response() +// sdk/rust/nym-sdk/src/ip_packet_client/discovery.rs — parse_connect_response() +/// Parse an IPR connect response, returning allocated IPs on success. +pub fn parse_connect_response(response: IpPacketResponse) -> Result { + let control_response = match response.data { + IpPacketResponseData::Control(c) => c, + other => return Err(IprResponseError::UnexpectedResponse(other)), + }; + + match *control_response { + ControlResponse::Connect(connect_resp) => match connect_resp.reply { + ConnectResponseReply::Success(success) => Ok(success.ips), + ConnectResponseReply::Failure(reason) => Err(IprResponseError::ConnectDenied(reason)), + }, + _ => Err(IprResponseError::UnexpectedResponse( + IpPacketResponseData::Control(control_response), + )), + } +} + +// Extracted from: +// nym-ip-packet-client/src/listener.rs — IprListener::handle_reconstructed_message() +// sdk/rust/nym-sdk/src/ip_packet_client/listener.rs — handle_ipr_response() +/// Parse raw IPR response bytes into an outcome. +/// +/// Logs non-fatal conditions (unknown control messages, deserialization +/// failures) and returns `None` for them. +pub fn handle_ipr_response(data: &[u8]) -> Option { + match IpPacketResponse::from_bytes(data) { + Ok(response) => match response.data { + IpPacketResponseData::Data(data_response) => { + let mut codec = MultiIpPacketCodec::new(); + let mut buf = BytesMut::from(data_response.ip_packet.as_ref()); + let mut packets = Vec::new(); + loop { + match codec.decode(&mut buf) { + Ok(Some(packet)) => packets.push(packet.into_bytes()), + Ok(None) => break, + Err(e) => { + warn!("Failed to decode bundled IP packet: {e}"); + break; + } + } + } + Some(MixnetMessageOutcome::IpPackets(packets)) + } + IpPacketResponseData::Control(control_response) => match *control_response { + ControlResponse::Connect(_) => { + info!("Received connect response when already connected - ignoring"); + None + } + ControlResponse::Disconnect(_) | ControlResponse::UnrequestedDisconnect(_) => { + info!("Received disconnect from IPR"); + Some(MixnetMessageOutcome::Disconnect) + } + ControlResponse::Pong(_) => { + info!("Received pong response"); + None + } + ControlResponse::Health(_) => { + info!("Received health response"); + None + } + ControlResponse::Info(info_resp) => { + let msg = format!( + "Received info response from the mixnet: {}", + info_resp.reply + ); + match info_resp.level { + InfoLevel::Info => info!("{msg}"), + InfoLevel::Warn => warn!("{msg}"), + InfoLevel::Error => error!("{msg}"), + } + None + } + }, + }, + Err(err) => { + warn!("Failed to deserialize IPR response: {err}"); + None + } + } +} diff --git a/common/ip-packet-requests/src/v8/response.rs b/common/ip-packet-requests/src/v8/response.rs index 12256252bd..1dce990caa 100644 --- a/common/ip-packet-requests/src/v8/response.rs +++ b/common/ip-packet-requests/src/v8/response.rs @@ -179,11 +179,15 @@ impl IpPacketResponse { make_bincode_serializer().serialize(self) } + pub fn from_bytes(data: &[u8]) -> Result { + use bincode::Options; + make_bincode_serializer().deserialize(data) + } + pub fn from_reconstructed_message( message: &nym_sphinx::receiver::ReconstructedMessage, ) -> Result { - use bincode::Options; - make_bincode_serializer().deserialize(&message.message) + Self::from_bytes(&message.message) } } diff --git a/common/ip-packet-requests/src/v9/mod.rs b/common/ip-packet-requests/src/v9/mod.rs new file mode 100644 index 0000000000..881a7c19d9 --- /dev/null +++ b/common/ip-packet-requests/src/v9/mod.rs @@ -0,0 +1,34 @@ +pub const VERSION: u8 = 9; + +/// Minimum nym-node release version that supports v9 (LP Stream framing). +/// Nodes running older versions will not understand LP-wrapped packets. +pub const MIN_RELEASE_VERSION: semver::Version = semver::Version::new(1, 30, 0); + +// v9 uses the same wire format as v8. The version bump indicates +// the message was sent with LP framing (SphinxStream). +// +// Types are re-exported for deserialization/matching. Use the wrapper +// constructors below to create correctly-versioned packets — never +// manually set `protocol.version` or `response.version`. +pub use super::v8::{request, response}; + +/// Create a v9 connect request (version byte set to 9). +pub fn new_connect_request(buffer_timeout: Option) -> (request::IpPacketRequest, u64) { + let (mut req, id) = request::IpPacketRequest::new_connect_request(buffer_timeout); + req.protocol.version = VERSION; + (req, id) +} + +/// Create a v9 data request (version byte set to 9). +pub fn new_data_request(data: bytes::Bytes) -> request::IpPacketRequest { + let mut req = request::IpPacketRequest::new_data_request(data); + req.protocol.version = VERSION; + req +} + +/// Create a v9 IP packet response (version byte set to 9). +pub fn new_ip_packet_response(ip_packet: bytes::Bytes) -> response::IpPacketResponse { + let mut resp = response::IpPacketResponse::new_ip_packet(ip_packet); + resp.version = VERSION; + resp +} diff --git a/nym-ip-packet-client/src/connect.rs b/nym-ip-packet-client/src/connect.rs index bf679f67cf..0448c0cf40 100644 --- a/nym-ip-packet-client/src/connect.rs +++ b/nym-ip-packet-client/src/connect.rs @@ -11,14 +11,10 @@ use tokio::time::sleep; use tokio_util::sync::CancellationToken; use tracing::{debug, error}; +use nym_ip_packet_requests::response_helpers::{self, IprResponseError}; + use crate::{ - current::{ - request::IpPacketRequest, - response::{ - ConnectResponse, ConnectResponseReply, ControlResponse, IpPacketResponse, - IpPacketResponseData, - }, - }, + current::{request::IpPacketRequest, response::IpPacketResponse}, error::{Error, Result}, helpers::check_ipr_message_version, }; @@ -101,32 +97,6 @@ impl IprClientConnect { Ok(request_id) } - async fn handle_connect_response(&self, response: ConnectResponse) -> Result { - debug!("Handling dynamic connect response"); - match response.reply { - ConnectResponseReply::Success(r) => Ok(r.ips), - ConnectResponseReply::Failure(reason) => Err(Error::ConnectRequestDenied { reason }), - } - } - - async fn handle_ip_packet_router_response(&self, response: IpPacketResponse) -> Result { - let control_response = match response.data { - IpPacketResponseData::Control(control_response) => control_response, - _ => { - error!("Received non-control response while waiting for connect response"); - return Err(Error::UnexpectedConnectResponse); - } - }; - - match *control_response { - ControlResponse::Connect(resp) => self.handle_connect_response(resp).await, - response => { - error!("Unexpected response: {response:?}"); - Err(Error::UnexpectedConnectResponse) - } - } - } - async fn listen_for_connect_response(&mut self, request_id: u64) -> Result { // Connecting is basically synchronous from the perspective of the mixnet client, so it's safe // to just grab ahold of the mutex and keep it until we get the response. @@ -173,7 +143,14 @@ impl IprClientConnect { if response.id() == Some(request_id) { tracing::debug!("Got response with matching id"); - return self.handle_ip_packet_router_response(response).await; + // Replaces local handle_ip_packet_router_response() + handle_connect_response() + return response_helpers::parse_connect_response(response) + .map_err(|e| match e { + IprResponseError::ConnectDenied(reason) => { + Error::ConnectRequestDenied { reason } + } + _ => Error::UnexpectedConnectResponse, + }); } } } diff --git a/nym-ip-packet-client/src/helpers.rs b/nym-ip-packet-client/src/helpers.rs index 8198a8a7b7..1b31870ff8 100644 --- a/nym-ip-packet-client/src/helpers.rs +++ b/nym-ip-packet-client/src/helpers.rs @@ -1,30 +1,24 @@ // Copyright 2023-2024 - Nym Technologies SA // SPDX-License-Identifier: GPL-3.0-only -use std::cmp::Ordering; - +use nym_ip_packet_requests::response_helpers::IprResponseError; use nym_sdk::mixnet::ReconstructedMessage; -use crate::{Error, current::VERSION as CURRENT_VERSION, error::Result}; +use crate::{current::VERSION as CURRENT_VERSION, error::Result}; pub(crate) fn check_ipr_message_version(message: &ReconstructedMessage) -> Result<()> { - // Assuming it's a IPR message, it will have a version as its first byte - if let Some(version) = message.message.first() { - match version.cmp(&CURRENT_VERSION) { - Ordering::Greater => Err(Error::ReceivedResponseWithNewVersion { - expected: CURRENT_VERSION, - received: *version, - }), - Ordering::Less => Err(Error::ReceivedResponseWithOldVersion { - expected: CURRENT_VERSION, - received: *version, - }), - Ordering::Equal => { - // We're good - Ok(()) - } + nym_ip_packet_requests::response_helpers::check_ipr_message_version( + &message.message, + CURRENT_VERSION, + ) + .map_err(|e| match e { + IprResponseError::NoVersionByte => crate::Error::NoVersionInMessage, + IprResponseError::VersionMismatch { expected, received } if received < expected => { + crate::Error::ReceivedResponseWithOldVersion { expected, received } } - } else { - Err(Error::NoVersionInMessage) - } + IprResponseError::VersionMismatch { expected, received } => { + crate::Error::ReceivedResponseWithNewVersion { expected, received } + } + _ => crate::Error::NoVersionInMessage, + }) } diff --git a/nym-ip-packet-client/src/lib.rs b/nym-ip-packet-client/src/lib.rs index 8aa5ffdf42..0ab5c2eb24 100644 --- a/nym-ip-packet-client/src/lib.rs +++ b/nym-ip-packet-client/src/lib.rs @@ -11,4 +11,4 @@ pub use error::Error; pub use listener::{IprListener, MixnetMessageOutcome}; // Re-export the currently used version -pub use nym_ip_packet_requests::v8 as current; +pub use nym_ip_packet_requests::v9 as current; diff --git a/sdk/rust/nym-sdk/Cargo.toml b/sdk/rust/nym-sdk/Cargo.toml index ded83e807a..9fddca8295 100644 --- a/sdk/rust/nym-sdk/Cargo.toml +++ b/sdk/rust/nym-sdk/Cargo.toml @@ -65,6 +65,9 @@ url = { workspace = true } toml = { workspace = true } tempfile = { workspace = true } +nym-ip-packet-requests = { workspace = true } +semver = { workspace = true } + # tcpproxy dependencies clap = { workspace = true, features = ["derive"] } anyhow.workspace = true @@ -94,6 +97,8 @@ tokio-stream = { workspace = true } tokio-util = { workspace = true, features = ["codec"] } parking_lot = { workspace = true } hex = { workspace = true } +pnet_packet = { workspace = true } +nym-ip-packet-requests = { workspace = true, features = ["test-utils"] } [features] libp2p-vanilla = [] diff --git a/sdk/rust/nym-sdk/examples/ipr_tunnel.rs b/sdk/rust/nym-sdk/examples/ipr_tunnel.rs new file mode 100644 index 0000000000..9bf242827e --- /dev/null +++ b/sdk/rust/nym-sdk/examples/ipr_tunnel.rs @@ -0,0 +1,92 @@ +// Copyright 2025 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +// Smoke test for IpMixStream: connect to an IPR, send a ping, check we get a reply. +// Tests both IPv4 and IPv6 paths. +// +// Usage: +// cargo run --example ipr_tunnel +// cargo run --example ipr_tunnel -- --ipr +// +// e.g. cargo run --example ipr_tunnel -- --ipr 6B6iuWX4bQP4GVA4Yq7XmZencaaGw6BaPY6xJWYSwsbF.6g6LRx1fgU2Q2A4ZPKonYHtfBARh1GPMe1LtXk6vpRR8@q2A2cbooyC16YJzvdYaSMH9X3cSiieZNtfBr8cE8Fi1 + +use std::net::{Ipv4Addr, Ipv6Addr}; +use std::time::Duration; + +use nym_ip_packet_requests::codec::MultiIpPacketCodec; +use nym_ip_packet_requests::icmp_utils::{ + build_icmp_ping, build_icmpv6_ping, is_echo_reply_v4, is_echo_reply_v6, +}; +use nym_sdk::ipr_wrapper::IpMixStream; + +const PING4_TARGET: Ipv4Addr = Ipv4Addr::new(8, 8, 8, 8); +const PING6_TARGET: Ipv6Addr = Ipv6Addr::new(0x2001, 0x4860, 0x4860, 0, 0, 0, 0, 0x8888); + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + nym_bin_common::logging::setup_tracing_logger(); + + let args: Vec = std::env::args().collect(); + let ipr_addr = args + .iter() + .position(|a| a == "--ipr") + .and_then(|i| args.get(i + 1)); + + let mut tunnel = if let Some(addr) = ipr_addr { + let recipient = addr.parse().expect("invalid IPR address"); + IpMixStream::new_with_ipr(recipient).await? + } else { + IpMixStream::new().await? + }; + + let ips = tunnel.allocated_ips(); + let src4 = ips.ipv4; + let src6 = ips.ipv6; + println!("Tunnel up — IPv4: {src4}, IPv6: {src6}"); + + // Send IPv4 ping (ICMP seq=0, unrelated to LP Stream sequence numbers) + let pkt4 = build_icmp_ping(src4, PING4_TARGET, 0).expect("failed to build ICMP packet"); + let bundled = MultiIpPacketCodec::bundle_one_packet(pkt4.into()); + tunnel.send_ip_packet(&bundled).await?; + println!("Sent ping → {PING4_TARGET}"); + + // Send IPv6 ping + let pkt6 = build_icmpv6_ping(src6, PING6_TARGET, 0).expect("failed to build ICMPv6 packet"); + let bundled = MultiIpPacketCodec::bundle_one_packet(pkt6.into()); + tunnel.send_ip_packet(&bundled).await?; + println!("Sent ping → {PING6_TARGET}"); + + let mut got_v4 = false; + let mut got_v6 = false; + let deadline = tokio::time::sleep(Duration::from_secs(30)); + tokio::pin!(deadline); + + loop { + tokio::select! { + _ = &mut deadline => { + if !got_v4 { println!("FAIL — no IPv4 reply within 30s"); } + if !got_v6 { println!("FAIL — no IPv6 reply within 30s"); } + break; + } + result = tunnel.handle_incoming() => { + for pkt in result? { + if !got_v4 && is_echo_reply_v4(&pkt, src4) { + println!("OK — got IPv4 echo reply"); + got_v4 = true; + } + if !got_v6 && is_echo_reply_v6(&pkt, src6) { + println!("OK — got IPv6 echo reply"); + got_v6 = true; + } + if got_v4 && got_v6 { + tunnel.disconnect().await; + return Ok(()); + } + } + } + } + } + + tunnel.disconnect().await; + Ok(()) +} diff --git a/sdk/rust/nym-sdk/src/error.rs b/sdk/rust/nym-sdk/src/error.rs index 57e486172a..dc2a5da492 100644 --- a/sdk/rust/nym-sdk/src/error.rs +++ b/sdk/rust/nym-sdk/src/error.rs @@ -1,3 +1,8 @@ +// Copyright 2025 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use nym_ip_packet_requests::v8::response::{ConnectFailureReason, IpPacketResponseData}; +use nym_validator_client::nym_api::error::NymAPIError; use nym_validator_client::nyxd::error::NyxdError; use std::path::PathBuf; @@ -108,6 +113,37 @@ pub enum Error { #[error("Stream subsystem failed to initialise: reconstructed_receiver unavailable")] StreamInitFailure, + + #[error("client not connected")] + IprStreamClientNotConnected, + + #[error("listening for connection response timed out")] + IPRConnectResponseTimeout, + + #[error("stream closed")] + IPRClientStreamClosed, + + #[error("expected control response, got {0:?}")] + UnexpectedResponseType(IpPacketResponseData), + + #[error("connect denied: {0:?}")] + ConnectDenied(ConnectFailureReason), + + #[allow(clippy::result_large_err)] + #[error("api directory error: {0}")] + GatewayDirectoryError(#[from] NymAPIError), + + #[error("did not receive Nym API URL")] + NoNymAPIUrl, + + #[error("no available gateway")] + NoGatewayAvailable, + + #[error("tunnel disconnected by IPR")] + IprTunnelDisconnected, + + #[error("message version check failed: {0}")] + IPRMessageVersionCheckFailed(String), } impl Error { diff --git a/sdk/rust/nym-sdk/src/ip_packet_client.rs b/sdk/rust/nym-sdk/src/ip_packet_client.rs new file mode 100644 index 0000000000..93a4dae694 --- /dev/null +++ b/sdk/rust/nym-sdk/src/ip_packet_client.rs @@ -0,0 +1,17 @@ +// Copyright 2025 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +//! IPR helpers for [`IpMixStream`](crate::ipr_wrapper::IpMixStream). +//! +//! - **Discovery** (`discovery`): queries the Nym API for IPR-enabled exit gateways. +//! - **Response handling** (`listener`): thin wrappers around +//! [`nym_ip_packet_requests::response_helpers`] that add version checking +//! and error mapping for SDK use. + +pub mod discovery; +pub mod listener; + +pub use listener::{handle_ipr_response, MixnetMessageOutcome}; + +// Re-export the currently used version +pub use nym_ip_packet_requests::v9 as current; diff --git a/sdk/rust/nym-sdk/src/ip_packet_client/discovery.rs b/sdk/rust/nym-sdk/src/ip_packet_client/discovery.rs new file mode 100644 index 0000000000..998567a951 --- /dev/null +++ b/sdk/rust/nym-sdk/src/ip_packet_client/discovery.rs @@ -0,0 +1,121 @@ +// Copyright 2025 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +//! IPR gateway discovery — find and rank IPR-enabled exit gateways via the Nym API. + +use std::collections::HashMap; + +use nym_crypto::asymmetric::ed25519; +use nym_ip_packet_requests::v9; +use nym_network_defaults::ApiUrl; +use nym_sphinx::addressing::clients::Recipient; +use nym_validator_client::nym_api::NymApiClientExt; +use tracing::{debug, error, info}; + +use rand::seq::SliceRandom; + +use crate::Error; + +#[derive(Clone)] +pub struct IprWithPerformance { + pub address: Recipient, + pub identity: ed25519::PublicKey, + pub performance: u8, +} + +#[allow(clippy::result_large_err)] +pub 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) +} + +pub async fn retrieve_exit_nodes_with_performance( + client: nym_http_api_client::Client, +) -> Result, Error> { + let all_nodes = client + .get_all_described_nodes_v2() + .await? + .into_iter() + .map(|described| (described.ed25519_identity_key(), described)) + .collect::>(); + + let exit_gateways = client.get_all_basic_nodes_with_metadata().await?.nodes; + + let mut described = Vec::new(); + + for exit in exit_gateways { + let Some(node) = all_nodes.get(&exit.ed25519_identity_pubkey) else { + continue; + }; + + // Only select nodes running a version that supports v9 (LP Stream framing) + let Ok(node_version) = semver::Version::parse(node.version()) else { + debug!( + "Skipping node {}: unable to parse version '{}'", + exit.ed25519_identity_pubkey, + node.version() + ); + continue; + }; + if node_version < v9::MIN_RELEASE_VERSION { + debug!( + "Skipping node {}: version {} < minimum {}", + exit.ed25519_identity_pubkey, + node_version, + v9::MIN_RELEASE_VERSION + ); + continue; + } + + if let Some(ipr_info) = node.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 the highest-performance IPR gateway from the directory. +pub async fn get_best_ipr(client: nym_http_api_client::Client) -> Result { + let nodes = retrieve_exit_nodes_with_performance(client).await?; + info!("Found {} Exit Gateways", nodes.len()); + + let selected_ipr = nodes + .choose_weighted(&mut rand::thread_rng(), |gw| gw.performance as f64) + .map_err(|_| Error::NoGatewayAvailable)?; + + let ipr_address = selected_ipr.address; + + info!( + "Using IPR: {} (Gateway: {}, Performance: {:?})", + ipr_address, selected_ipr.identity, selected_ipr.performance + ); + + Ok(ipr_address) +} diff --git a/sdk/rust/nym-sdk/src/ip_packet_client/listener.rs b/sdk/rust/nym-sdk/src/ip_packet_client/listener.rs new file mode 100644 index 0000000000..28ab5cc61a --- /dev/null +++ b/sdk/rust/nym-sdk/src/ip_packet_client/listener.rs @@ -0,0 +1,20 @@ +// Copyright 2025 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +pub use nym_ip_packet_requests::response_helpers::MixnetMessageOutcome; + +use nym_ip_packet_requests::response_helpers; + +use crate::ip_packet_client::current::VERSION as CURRENT_VERSION; + +/// Check that the first byte of an IPR message matches the expected protocol version. +pub(crate) fn check_ipr_message_version(data: &[u8]) -> Result<(), crate::Error> { + response_helpers::check_ipr_message_version(data, CURRENT_VERSION) + .map_err(|e| crate::Error::IPRMessageVersionCheckFailed(e.to_string())) +} + +/// Parse raw IPR response bytes into an outcome. +pub fn handle_ipr_response(data: &[u8]) -> Result, crate::Error> { + check_ipr_message_version(data)?; + Ok(response_helpers::handle_ipr_response(data)) +} diff --git a/sdk/rust/nym-sdk/src/ipr_wrapper.rs b/sdk/rust/nym-sdk/src/ipr_wrapper.rs new file mode 100644 index 0000000000..084b74bda0 --- /dev/null +++ b/sdk/rust/nym-sdk/src/ipr_wrapper.rs @@ -0,0 +1,13 @@ +// Copyright 2025 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +//! High-level IPR (IP Packet Router) stream wrapper. +//! +//! [`IpMixStream`] tunnels IP packets through the Nym mixnet to an exit +//! gateway running an IP Packet Router. Both requests and responses are +//! wrapped in LP Stream frames for type-safe detection at the IPR and +//! dispatch by the client's stream router. + +mod ip_mix_stream; + +pub use ip_mix_stream::IpMixStream; diff --git a/sdk/rust/nym-sdk/src/ipr_wrapper/architecture.md b/sdk/rust/nym-sdk/src/ipr_wrapper/architecture.md new file mode 100644 index 0000000000..0c19e01527 --- /dev/null +++ b/sdk/rust/nym-sdk/src/ipr_wrapper/architecture.md @@ -0,0 +1,80 @@ +# IpMixStream Architecture + +## Overview + +`IpMixStream` tunnels IP packets through the Nym mixnet to an IP Packet Router +(IPR) exit gateway. It provides a high-level API over a single `MixnetStream`, +which handles LP Stream framing and Sphinx packet transport automatically. + +## Data Flow + +```text +Client IPR + | | + |-- IpPacketRequest (connect) ---- LP Stream ----->| + |<--- IpPacketResponse (ips) ---- LP Stream (s=0) -| + | | + |-- IpPacketRequest (data) ------- LP Stream ----->| -> TUN -> internet + |<--- IpPacketResponse (data) --- LP Stream (s=1+) | <- TUN <- internet +``` + +## Layer Stack + +```text +IpMixStream IPR protocol (connect, data, disconnect) +MixnetStream AsyncRead + AsyncWrite, LP Stream framing, seq numbers +Stream Router Dispatches inbound messages by stream_id +MixnetClient Sphinx packet encryption, SURB management +Mixnet Entry GW -> Mix1 -> Mix2 -> Mix3 -> Exit GW +``` + +## LP Stream Framing + +All messages between client and IPR are wrapped in LP Stream frames: + +- **Client -> IPR**: `MixnetStream.write()` wraps each write in an LP Stream + Data frame (stream_id, sequence number, payload). The IPR detects + `LpFrameKind::Stream` and strips the header before processing. + +- **IPR -> Client**: Both inline responses (connect handshake, pong) and async + TUN responses are wrapped in LP Stream frames with the same stream_id. The + client's stream router dispatches by stream_id to the correct `MixnetStream`. + +## IP Allocation + +The IPR owns a subnet (`10.0.0.0/16` for IPv4, `fc00::/112` for IPv6) and +allocates addresses to clients on connect. Two modes are supported: + +- **Dynamic** (used by `IpMixStream`): The IPR picks a random unused `IpPair` + from the pool (up to 100 retries). `IpMixStream` uses this path. +- **Static**: The client requests specific IPs; the IPR checks availability. + +On dynamic connect: + +1. IPR calls `find_new_ip()` — random selection from the subnet +2. Registers the client in dual `HashMap` / `HashMap` + maps (keyed by allocated IP). The client identity (anonymous sender tag + for v8, Nym address for v6/v7) is stored inside the `ConnectedClient` value. +3. Returns `DynamicConnectSuccess { ips }` to the client + +The client uses the allocated IPv4/IPv6 as its source address when constructing +packets. When the TUN device receives a response destined for that IP, the IPR +looks up which client owns it and forwards the packet back through the mixnet. + +Reserved addresses: `10.0.0.1` / `fc00::1` (TUN gateway, never allocated). + +> **Note — pool isolation from LP dVPN:** This pool (`10.0.0.0/16`, `fc00::/112` +> on `nymtun`) is separate from the authenticator/LP dVPN WireGuard pool +> (`10.1.0.0/16`, `fc01::/112` on `nymwg`). The second octet differs (`10.0` vs +> `10.1`) and each uses its own TUN interface. Both sets of constants live in +> `common/network-defaults/src/constants.rs` (`mixnet_vpn` vs `wireguard` modules). +> The IPR addresses are hardcoded; the WG addresses are configurable but default to +> the non-overlapping range. + +## Connection Lifecycle + +1. `IpMixStream::new()` discovers the best IPR via Nym API +2. Opens a `MixnetStream` to the IPR (`client.open_stream()`) +3. Sends a dynamic connect request, IPR allocates an `IpPair` +4. Ready for `send_ip_packet()` / `handle_incoming()` loop +5. `disconnect()` shuts down the mixnet client diff --git a/sdk/rust/nym-sdk/src/ipr_wrapper/ip_mix_stream.rs b/sdk/rust/nym-sdk/src/ipr_wrapper/ip_mix_stream.rs new file mode 100644 index 0000000000..13cfb56e23 --- /dev/null +++ b/sdk/rust/nym-sdk/src/ipr_wrapper/ip_mix_stream.rs @@ -0,0 +1,192 @@ +// Copyright 2025 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 +use crate::ip_packet_client::{ + discovery::{create_nym_api_client, get_best_ipr}, + handle_ipr_response, + listener::check_ipr_message_version, + MixnetMessageOutcome, +}; +use crate::mixnet::{MixnetClient, MixnetStream, Recipient}; +use crate::Error; +use bytes::Bytes; +use nym_ip_packet_requests::response_helpers; +use nym_ip_packet_requests::{ + v9::{self, response::IpPacketResponse}, + IpPair, +}; +use nym_network_defaults::NymNetworkDetails; +use std::time::Duration; +use tokio::io::AsyncWriteExt; +use tracing::{debug, info}; + +const IPR_CONNECT_TIMEOUT: Duration = Duration::from_secs(60); + +/// A bidirectional tunnel for sending and receiving IP packets through the mixnet. +/// +/// Wraps a [`MixnetStream`] (opened to an IPR exit gateway) and provides a +/// high-level API for the IPR protocol. The underlying `MixnetStream` handles +/// LP Stream framing and stream multiplexing automatically. +/// +/// # Data flow +/// +/// ```text +/// IpMixStream.send_ip_packet(bytes) +/// → IpPacketRequest::to_bytes() → MixnetStream.write() +/// → LP Stream frame (stream_id, seq, Data) +/// → Sphinx packets → mixnet → IPR +/// +/// IPR processes request → TUN → internet → response +/// → IPR wraps in LP Stream frame → Sphinx → mixnet → client +/// → stream router dispatches by stream_id +/// → MixnetStream.recv() → IpPacketResponse bytes +/// → handle_ipr_response() → extract IP packets +/// ``` +pub struct IpMixStream { + stream: MixnetStream, + client: MixnetClient, + allocated_ips: IpPair, + connected: bool, +} + +impl IpMixStream { + /// Discover the best IPR, connect through the mixnet, and establish the IP tunnel. + /// + /// Returns a ready-to-use tunnel with allocated IP addresses. + pub async fn new() -> Result { + let network_defaults = NymNetworkDetails::new_mainnet(); + let api_client = + create_nym_api_client(network_defaults.nym_api_urls.ok_or(Error::NoNymAPIUrl)?)?; + let ipr_address = get_best_ipr(api_client).await?; + Self::new_with_ipr(ipr_address).await + } + + /// Connect to a specific IPR address. + /// + /// Use this when you already know the IPR `Recipient` address (e.g. for + /// testing against a specific exit node). For automatic discovery, use + /// [`IpMixStream::new`] instead. + pub async fn new_with_ipr(ipr_address: Recipient) -> Result { + nym_network_defaults::setup_env(None::<&str>); + let mut client = MixnetClient::connect_new().await?; + let mut stream = client.open_stream(ipr_address, Some(10)).await?; + + info!("Connecting to IP packet router at {ipr_address}"); + let allocated_ips = Self::connect_tunnel(&mut stream).await?; + info!( + "Connected — IPv4: {}, IPv6: {}", + allocated_ips.ipv4, allocated_ips.ipv6 + ); + + Ok(Self { + stream, + client, + allocated_ips, + connected: true, + }) + } + + pub fn nym_address(&self) -> &Recipient { + self.client.nym_address() + } + + pub fn allocated_ips(&self) -> &IpPair { + &self.allocated_ips + } + + pub fn is_connected(&self) -> bool { + self.connected + } + + /// Check that the tunnel is connected, returning an error if not. + pub fn check_connected(&self) -> Result<(), Error> { + if self.connected { + Ok(()) + } else { + Err(Error::IprStreamClientNotConnected) + } + } + + async fn connect_tunnel(stream: &mut MixnetStream) -> Result { + let (request, request_id) = v9::new_connect_request(None); + debug!("Sending connect request with ID: {}", request_id); + + let request_bytes = request.to_bytes()?; + stream + .write_all(&request_bytes) + .await + .map_err(|_| Error::MessageSendingFailure)?; + + let timeout = tokio::time::sleep(IPR_CONNECT_TIMEOUT); + tokio::pin!(timeout); + + loop { + tokio::select! { + _ = &mut timeout => { + return Err(Error::IPRConnectResponseTimeout); + } + result = stream.recv() => { + let data = result.ok_or(Error::IPRClientStreamClosed)?; + + check_ipr_message_version(&data)?; + if let Ok(response) = IpPacketResponse::from_bytes(&data) { + if response.id() == Some(request_id) { + return response_helpers::parse_connect_response(response) + .map_err(|e| match e { + response_helpers::IprResponseError::ConnectDenied(r) => Error::ConnectDenied(r), + response_helpers::IprResponseError::UnexpectedResponse(d) => Error::UnexpectedResponseType(d), + other => Error::IPRMessageVersionCheckFailed(other.to_string()), + }); + } + } + } + } + } + } + + /// Send an IP packet through the tunnel. + pub async fn send_ip_packet(&mut self, packet: &[u8]) -> Result<(), Error> { + self.check_connected()?; + let request = v9::new_data_request(packet.to_vec().into()); + let request_bytes = request.to_bytes()?; + self.stream + .write_all(&request_bytes) + .await + .map_err(|_| Error::MessageSendingFailure) + } + + /// Handle incoming messages from the mixnet. + /// + /// Reads from the underlying `MixnetStream`, parses IPR responses, and + /// extracts IP packets. Returns an empty vec on timeout (10 s). + pub async fn handle_incoming(&mut self) -> Result, Error> { + let data = match tokio::time::timeout(Duration::from_secs(10), self.stream.recv()).await { + Err(_) => return Ok(Vec::new()), + Ok(None) => { + self.connected = false; + return Err(Error::IPRClientStreamClosed); + } + Ok(Some(data)) => data, + }; + + match handle_ipr_response(&data) { + Ok(Some(MixnetMessageOutcome::IpPackets(packets))) => { + debug!("Extracted {} IP packets", packets.len()); + Ok(packets) + } + Ok(Some(MixnetMessageOutcome::Disconnect)) => { + info!("Received disconnect"); + self.connected = false; + Err(Error::IprTunnelDisconnected) + } + Ok(None) => Ok(Vec::new()), + Err(e) => Err(e), + } + } + + /// Disconnect from the Mixnet. Disconnected clients cannot be reconnected. + pub async fn disconnect(self) { + debug!("Disconnecting"); + self.client.disconnect().await; + debug!("Disconnected"); + } +} diff --git a/sdk/rust/nym-sdk/src/lib.rs b/sdk/rust/nym-sdk/src/lib.rs index 8b5bb45568..03fc5ccdbe 100644 --- a/sdk/rust/nym-sdk/src/lib.rs +++ b/sdk/rust/nym-sdk/src/lib.rs @@ -1,13 +1,19 @@ +// Copyright 2025 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + //! Rust SDK for the Nym platform //! //! The main component currently is [`mixnet`]. //! [`tcp_proxy`] is probably a good place to start for anyone wanting to integrate with existing app code and read/write from a socket. //! [`client_pool`] is a configurable client pool. +//! [`ipr_wrapper`] tunnels IP packets through the mixnet to an IPR exit gateway. mod error; pub mod bandwidth; pub mod client_pool; +pub mod ip_packet_client; +pub mod ipr_wrapper; pub mod mixnet; pub mod tcp_proxy; diff --git a/sdk/rust/nym-sdk/src/mixnet/stream/mixnet_stream.rs b/sdk/rust/nym-sdk/src/mixnet/stream/mixnet_stream.rs index 5ba1d08853..fd8a9c879f 100644 --- a/sdk/rust/nym-sdk/src/mixnet/stream/mixnet_stream.rs +++ b/sdk/rust/nym-sdk/src/mixnet/stream/mixnet_stream.rs @@ -1,3 +1,6 @@ +// Copyright 2025 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + //! Per-stream handle implementing `AsyncRead + AsyncWrite`. use std::pin::Pin; @@ -111,6 +114,17 @@ impl MixnetStream { self.id } + /// Receive a single message payload directly from the stream channel. + /// + /// Returns `None` on EOF (channel closed). Drains any leftover from + /// a prior `AsyncRead` call first. + pub async fn recv(&mut self) -> Option> { + if !self.read_buf.is_empty() { + return Some(self.read_buf.split().to_vec()); + } + self.inbound_rx.recv().await + } + /// Wrap `data` in the appropriate `InputMessage` for this stream's destination. fn make_input_message(&self, data: Vec) -> InputMessage { match &self.destination { diff --git a/sdk/rust/nym-sdk/src/mixnet/stream/mod.rs b/sdk/rust/nym-sdk/src/mixnet/stream/mod.rs index 2e648099a9..d3e85bd800 100644 --- a/sdk/rust/nym-sdk/src/mixnet/stream/mod.rs +++ b/sdk/rust/nym-sdk/src/mixnet/stream/mod.rs @@ -1,3 +1,6 @@ +// Copyright 2025 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + //! Stream multiplexing for `MixnetClient`. //! //! A [`MixnetStream`] is a byte channel (`AsyncRead + AsyncWrite`) to a @@ -9,12 +12,12 @@ //! stream's channel (or to the listener for `Open` messages). mod mixnet_stream; -mod protocol; +pub(crate) mod protocol; pub use mixnet_stream::MixnetStream; pub use protocol::StreamId; -use std::collections::HashMap; +use std::collections::{BTreeMap, HashMap}; use std::sync::atomic::Ordering; use std::sync::Arc; use std::time::Duration; @@ -47,11 +50,25 @@ pub(crate) const DEFAULT_STREAM_IDLE_TIMEOUT: Duration = Duration::from_secs(30 const MAX_CLEANUP_INTERVAL: Duration = Duration::from_secs(10); /// Per-stream state stored in the routing table. +/// +/// Reorder buffer uses the same BTreeMap pattern as `OrderedMessageBuffer` +/// (`common/socks5/ordered-buffer/`) but drains per-message instead of +/// concatenating, so `recv()` preserves message boundaries. struct StreamEntry { sender: mpsc::UnboundedSender>, last_activity: Instant, + next_seq: u32, + pending: BTreeMap>, } +/// Maximum number of out-of-order messages buffered per stream before we +/// skip ahead. Without this cap, a malicious sender that deliberately skips +/// a sequence number (e.g. never sends seq 1) could cause the buffer to +/// grow indefinitely while the drain loop waits for the missing seq. +/// The idle timeout only reaps *inactive* streams, so an actively-sending +/// attacker would bypass it. +const MAX_REORDER_BUFFER: usize = 256; + /// The shared stream routing table. /// /// Wraps the map of active streams behind an async mutex with focused @@ -76,6 +93,8 @@ impl StreamMap { StreamEntry { sender: tx, last_activity: Instant::now(), + next_seq: 0, + pending: BTreeMap::new(), }, ); rx @@ -95,17 +114,49 @@ impl StreamMap { }); } - /// 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) { + /// Buffer a message and flush any contiguous sequence to the channel. + /// Updates `last_activity` on success; removes the entry if the + /// receiver has been dropped. + async fn send_to_stream(&self, stream_id: &StreamId, seq: u32, 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 + if seq < entry.next_seq { + warn!( + "Stream {stream_id}: dropping old seq {seq} (expected >= {})", + entry.next_seq + ); } else { - entry.last_activity = Instant::now(); - false + entry.pending.insert(seq, data); } + + // If the buffer has grown too large, skip ahead to the lowest + // buffered seq so we don't accumulate unbounded memory. + if entry.pending.len() > MAX_REORDER_BUFFER { + if let Some(&lowest) = entry.pending.keys().next() { + warn!( + "Stream {stream_id}: reorder buffer overflow ({} pending), \ + skipping seq {} -> {lowest}", + entry.pending.len(), + entry.next_seq + ); + entry.next_seq = lowest; + } + } + + // Drain contiguous messages + let mut failed = false; + while let Some(msg) = entry.pending.remove(&entry.next_seq) { + if entry.sender.send(msg).is_err() { + failed = true; + break; + } + entry.next_seq += 1; + } + + if !failed { + entry.last_activity = Instant::now(); + } + failed } else { false }; @@ -242,7 +293,7 @@ async fn run_router( } SphinxStreamMsgType::Data => { streams - .send_to_stream(&stream_id, frame.data.to_vec()) + .send_to_stream(&stream_id, frame.sequence_num, frame.data.to_vec()) .await; } } @@ -295,10 +346,9 @@ pub(crate) async fn open_stream( let stream_id = StreamId::random(); let rx = streams.register_stream(stream_id).await; - // Currently hardcoded as we don't have message ordering in place *yet* in the SDK - // streams - when it *is* added then it will set the receiver's expected starting seq - // number. Gives us the ability down the road to e.g. pick up a dropped stream from - // where it left off. + // Open message with seq=0. The receiver's reorder buffer starts at + // next_seq=0 so this could later carry an initial seq to resume a + // dropped stream from where it left off. let wire = encode_stream_message(&stream_id, SphinxStreamMsgType::Open, 0, &[]); let msg = InputMessage::new_anonymous( recipient, @@ -379,7 +429,7 @@ mod tests { 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; + map.send_to_stream(&id, 0, vec![1, 2, 3]).await; // Advance past the original timeout, but only 5s since last activity tokio::time::advance(Duration::from_secs(5)).await; @@ -405,4 +455,37 @@ mod tests { assert_eq!(map.inner.lock().await.len(), 1); } + + #[tokio::test] + async fn out_of_order_messages_delivered_in_sequence() { + let map = StreamMap::new(); + let id = StreamId::random(); + let mut rx = map.register_stream(id).await; + + // Send seq 2, 0, 1 out of order + map.send_to_stream(&id, 2, vec![20]).await; + map.send_to_stream(&id, 0, vec![0]).await; + + // seq 0 should be delivered now, but 2 is buffered (gap at 1) + assert_eq!(rx.recv().await.unwrap(), vec![0]); + + // Fill the gap — both 1 and 2 should flush + map.send_to_stream(&id, 1, vec![10]).await; + assert_eq!(rx.recv().await.unwrap(), vec![10]); + assert_eq!(rx.recv().await.unwrap(), vec![20]); + } + + #[tokio::test] + async fn duplicate_seq_is_dropped() { + let map = StreamMap::new(); + let id = StreamId::random(); + let mut rx = map.register_stream(id).await; + + map.send_to_stream(&id, 0, vec![0]).await; + map.send_to_stream(&id, 0, vec![99]).await; // duplicate, dropped + map.send_to_stream(&id, 1, vec![1]).await; + + assert_eq!(rx.recv().await.unwrap(), vec![0]); + assert_eq!(rx.recv().await.unwrap(), vec![1]); + } } diff --git a/sdk/rust/nym-sdk/src/mixnet/stream/protocol.rs b/sdk/rust/nym-sdk/src/mixnet/stream/protocol.rs index 68e34176fc..058e6c445d 100644 --- a/sdk/rust/nym-sdk/src/mixnet/stream/protocol.rs +++ b/sdk/rust/nym-sdk/src/mixnet/stream/protocol.rs @@ -1,3 +1,6 @@ +// Copyright 2025 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + //! Wire protocol for stream multiplexing. //! //! Every message between streams carries an LP frame header prepended to @@ -53,7 +56,6 @@ impl fmt::Display for StreamId { pub struct StreamFrame<'a> { pub stream_id: StreamId, pub msg_type: SphinxStreamMsgType, - #[allow(dead_code)] // will be used for reordering pub sequence_num: u32, pub data: &'a [u8], } diff --git a/service-providers/ip-packet-router/Cargo.toml b/service-providers/ip-packet-router/Cargo.toml index 5e3340871f..fdb12ae514 100644 --- a/service-providers/ip-packet-router/Cargo.toml +++ b/service-providers/ip-packet-router/Cargo.toml @@ -26,7 +26,7 @@ nym-crypto = { workspace = true } nym-exit-policy = { workspace = true } nym-id = { workspace = true } nym-ip-packet-requests = { workspace = true } -nym-kcp = { path = "../../common/nym-kcp" } # TODO MAX add to workspace dependencies +nym-lp = { workspace = true } nym-network-defaults = { workspace = true } nym-network-requester = { path = "../network-requester" } nym-sdk = { workspace = true } diff --git a/service-providers/ip-packet-router/src/clients/connected_client_handler.rs b/service-providers/ip-packet-router/src/clients/connected_client_handler.rs index ba9f9b3290..e44b543276 100644 --- a/service-providers/ip-packet-router/src/clients/connected_client_handler.rs +++ b/service-providers/ip-packet-router/src/clients/connected_client_handler.rs @@ -1,6 +1,7 @@ // Copyright 2025 - Nym Technologies SA // SPDX-License-Identifier: GPL-3.0-only +use std::sync::atomic::{AtomicU32, Ordering}; use std::time::Duration; use bytes::{Bytes, BytesMut}; @@ -9,6 +10,10 @@ use nym_ip_packet_requests::{ v6::response::IpPacketResponse as IpPacketResponseV6, v7::response::IpPacketResponse as IpPacketResponseV7, v8::response::IpPacketResponse as IpPacketResponseV8, + v9, +}; +use nym_lp::packet::frame::{ + LpFrame, LpFrameHeader, SphinxStreamFrameAttributes, SphinxStreamMsgType, }; use nym_sdk::mixnet::{ InputMessage, MixnetClientSender, MixnetMessageSender, MixnetMessageSinkTranslator, @@ -26,6 +31,19 @@ use crate::{ messages::ClientVersion, }; +/// Encode a payload inside an LP Stream frame: `[LpFrameHeader][payload]`. +pub(crate) fn encode_stream_frame(stream_id: u64, sequence_num: u32, payload: Vec) -> Vec { + let attrs = SphinxStreamFrameAttributes { + stream_id, + msg_type: SphinxStreamMsgType::Data, + sequence_num, + }; + let frame = LpFrame::new_stream(attrs, payload); + let mut buf = BytesMut::with_capacity(LpFrameHeader::SIZE + frame.content.len()); + frame.encode(&mut buf); + buf.to_vec() +} + // Data flow // Out: mixnet_listener -> decode -> handle_packet -> write_to_tun // In: tun_listener -> [connected_client_handler -> encode] -> mixnet_sender @@ -64,6 +82,7 @@ impl ConnectedClientHandler { buffer_timeout: Duration, client_version: ClientVersion, mixnet_client_sender: MixnetClientSender, + stream_id: Option, ) -> ( mpsc::UnboundedSender>, oneshot::Sender<()>, @@ -71,6 +90,9 @@ impl ConnectedClientHandler { ) { log::debug!("Starting connected client handler for: {client_id}"); log::debug!("client version: {client_version:?}"); + if let Some(sid) = stream_id { + log::debug!("LP Stream mode: stream_id={sid:#018x}"); + } let (close_tx, close_rx) = oneshot::channel(); let (forward_from_tun_tx, forward_from_tun_rx) = mpsc::unbounded_channel(); @@ -86,6 +108,13 @@ impl ConnectedClientHandler { let input_message_creator = ToIprDataResponse { send_to: client_id.clone(), client_version, + stream_id, + // Start at 1 — seq=0 is reserved for inline control responses + // (connect handshake, pong, health) sent by + // handle_stream_response() in mixnet_listener.rs. + // The skip-on-wrap logic in to_input_message() ensures we never + // emit seq=0 from the data path either. + next_response_seq: AtomicU32::new(1), }; let connected_client_handler = ConnectedClientHandler { @@ -189,15 +218,21 @@ fn create_ip_packet_response( ClientVersion::V6 => IpPacketResponseV6::new_ip_packet(packets).to_bytes(), ClientVersion::V7 => IpPacketResponseV7::new_ip_packet(packets).to_bytes(), ClientVersion::V8 => IpPacketResponseV8::new_ip_packet(packets).to_bytes(), + ClientVersion::V9 => v9::new_ip_packet_response(packets).to_bytes(), } } -// This struct is used by the sink to translate the the bundled IP packets into a IPR packet -// responses that can be sent to the mixnet. -#[derive(Clone, Debug)] +// This struct is used by the sink to translate the bundled IP packets into IPR packet +// responses that can be sent to the mixnet. When `stream_id` is set, responses are +// wrapped in LP Stream frames so the client's stream router can dispatch them. +#[derive(Debug)] struct ToIprDataResponse { send_to: ConnectedClientId, client_version: ClientVersion, + /// When Some, wrap responses in LP Stream frames with this stream_id. + stream_id: Option, + /// Sequence number for LP Stream response frames. + next_response_seq: AtomicU32, } impl MixnetMessageSinkTranslator for ToIprDataResponse { @@ -205,12 +240,24 @@ impl MixnetMessageSinkTranslator for ToIprDataResponse { &self, bundled_ip_packets: &[u8], ) -> std::result::Result { - // Create a IPR packet response that the recipient can understand let response_packet = create_ip_packet_response(bundled_ip_packets, self.client_version)?; - // Wrap the response packet in a mixnet input message + // Optionally wrap in LP Stream frame for stream-mode clients. + // Seq counter starts at 1 and skips 0 on wrap-around because seq=0 + // is used by handle_stream_response() for inline control responses + // (see comment in mixnet_listener.rs for the full story). + let final_packet = if let Some(stream_id) = self.stream_id { + let mut seq = self.next_response_seq.fetch_add(1, Ordering::Relaxed); + if seq == 0 { + seq = self.next_response_seq.fetch_add(1, Ordering::Relaxed); + } + encode_stream_frame(stream_id, seq, response_packet) + } else { + response_packet + }; + let input_message = - crate::util::create_message::create_input_message(&self.send_to, response_packet) + crate::util::create_message::create_input_message(&self.send_to, final_packet) .with_max_retransmissions(0); Ok(input_message) @@ -282,6 +329,8 @@ mod tests { let bytes_to_input_message = ToIprDataResponse { send_to: client_id.clone(), client_version, + stream_id: None, + next_response_seq: AtomicU32::new(0), }; let mixnet_ip_packet_sender = MixnetMessageSink::new_with_custom_translator( diff --git a/service-providers/ip-packet-router/src/clients/mod.rs b/service-providers/ip-packet-router/src/clients/mod.rs index fc6b39a359..1be30a4fa9 100644 --- a/service-providers/ip-packet-router/src/clients/mod.rs +++ b/service-providers/ip-packet-router/src/clients/mod.rs @@ -6,7 +6,7 @@ mod connected_client_handler; mod connected_clients; pub(crate) use client_id::ConnectedClientId; -pub(crate) use connected_client_handler::ConnectedClientHandler; +pub(crate) use connected_client_handler::{ConnectedClientHandler, encode_stream_frame}; pub(crate) use connected_clients::{ ConnectEvent, ConnectedClientEvent, ConnectedClients, DisconnectEvent, }; diff --git a/service-providers/ip-packet-router/src/error.rs b/service-providers/ip-packet-router/src/error.rs index b77b641455..ed1ca47263 100644 --- a/service-providers/ip-packet-router/src/error.rs +++ b/service-providers/ip-packet-router/src/error.rs @@ -117,8 +117,8 @@ pub enum IpPacketRouterError { #[error("failed to deserialize protocol: {source}")] FailedToDeserializeProtocol { source: ProtocolError }, - #[error("KCP protocol error: {0}")] - KcpError(String), + #[error("{0}")] + Other(String), } pub type Result = std::result::Result; diff --git a/service-providers/ip-packet-router/src/ip_packet_router.rs b/service-providers/ip-packet-router/src/ip_packet_router.rs index 084adb4009..13d5a7f6a8 100644 --- a/service-providers/ip-packet-router/src/ip_packet_router.rs +++ b/service-providers/ip-packet-router/src/ip_packet_router.rs @@ -180,7 +180,6 @@ impl IpPacketRouter { mixnet_client, shutdown_token: self.shutdown.clone_shutdown_token(), connected_clients, - kcp_session_manager: crate::kcp_session_manager::KcpSessionManager::new(), }; log::info!("The address of this client is: {self_address}"); diff --git a/service-providers/ip-packet-router/src/kcp_session_manager.rs b/service-providers/ip-packet-router/src/kcp_session_manager.rs deleted file mode 100644 index 9175f5d44e..0000000000 --- a/service-providers/ip-packet-router/src/kcp_session_manager.rs +++ /dev/null @@ -1,504 +0,0 @@ -// Copyright 2025 - Nym Technologies SA -// SPDX-License-Identifier: GPL-3.0-only - -#![allow(unused)] - -//! KCP Session Manager for LP clients at the exit gateway. -//! -//! This module sits between Sphinx unwrapping and IPR message processing. -//! It maintains per-client KCP state (keyed by conv_id from KCP packets), -//! reassembles KCP fragments into complete messages, and wraps responses -//! in KCP for SURB replies. -//! -//! # Architecture -//! -//! ```text -//! Mixnet → [Sphinx unwrap] → [KCP Session Manager] → [IPR message handling] -//! ↓ -//! KCP sessions per conv_id -//! ↓ -//! Reassemble fragments → DataRequest -//! ``` -//! -//! # Design Notes -//! -//! - Conv ID is extracted from the first 4 bytes of KCP packet data -//! - SURBs are stored per conv_id for sending replies -//! - Pattern follows `nym-lp-node::Node` from lewes-protocol - -use bytes::BytesMut; -use nym_kcp::driver::KcpDriver; -use nym_kcp::packet::KcpPacket; -use nym_kcp::session::KcpSession; -use nym_sphinx::anonymous_replies::ReplySurb; -use nym_sphinx::anonymous_replies::requests::AnonymousSenderTag; -use std::collections::{HashMap, VecDeque}; -use std::time::{Duration, Instant}; - -use crate::error::IpPacketRouterError; - -/// Default session timeout (5 minutes, matching IPR client timeout) -const SESSION_TIMEOUT: Duration = Duration::from_secs(300); - -/// Maximum concurrent KCP sessions per exit gateway -const MAX_SESSIONS: usize = 10000; - -/// State for a single KCP session -struct KcpSessionState { - driver: KcpDriver, - /// SURBs for sending replies back to this client - surbs: VecDeque, - /// Last activity timestamp - last_activity: Instant, - /// The sender tag associated with this session (for logging/debugging) - sender_tag: Option, -} - -impl KcpSessionState { - fn new(conv_id: u32) -> Self { - let session = KcpSession::new(conv_id); - Self { - driver: KcpDriver::new(session), - surbs: VecDeque::new(), - last_activity: Instant::now(), - sender_tag: None, - } - } - - fn touch(&mut self) { - self.last_activity = Instant::now(); - } - - fn is_expired(&self, timeout: Duration) -> bool { - self.last_activity.elapsed() > timeout - } - - fn add_surbs(&mut self, surbs: Vec) { - self.surbs.extend(surbs); - } - - fn take_surb(&mut self) -> Option { - self.surbs.pop_front() - } - - fn surb_count(&self) -> usize { - self.surbs.len() - } -} - -/// KCP Session Manager maintains per-client KCP state for LP clients. -/// -/// It intercepts incoming Sphinx payloads containing KCP data, extracts KCP frames, -/// reassembles them into complete messages, and returns the assembled data for IPR processing. -/// -/// Sessions are keyed by `conv_id` (first 4 bytes of KCP packet), which is derived -/// by clients from their local and remote addresses. -pub struct KcpSessionManager { - /// KCP sessions keyed by conv_id (from KCP packet header) - sessions: HashMap, - /// Session timeout duration - timeout: Duration, - /// Maximum number of sessions - max_sessions: usize, -} - -/// Result of processing incoming KCP data from a client -pub(crate) struct KcpProcessingResult { - /// The conv_id extracted from the KCP packet - pub(crate) conversation_id: u32, - - /// Vector of decoded KCP packets (for inspection/logging) - pub(crate) decoded_packets: Vec, - - /// Vector of complete reassembled messages ready for IPR processing - pub(crate) reassembled_messages: Vec>, -} - -impl Default for KcpSessionManager { - fn default() -> Self { - Self::new() - } -} - -impl KcpSessionManager { - /// Create a new KCP Session Manager with default settings - pub fn new() -> Self { - Self { - sessions: HashMap::new(), - timeout: SESSION_TIMEOUT, - max_sessions: MAX_SESSIONS, - } - } - - /// Create a new KCP Session Manager with custom settings - pub fn with_config(timeout: Duration, max_sessions: usize) -> Self { - Self { - sessions: HashMap::new(), - timeout, - max_sessions, - } - } - - /// Process incoming KCP data from a client. - /// - /// Takes raw KCP-encoded data (from a RepliableMessage). The conv_id is extracted - /// from the first 4 bytes of the KCP data. - /// - /// # Arguments - /// * `kcp_data` - Raw KCP packet data (conv_id is in first 4 bytes) - /// * `reply_surbs` - SURBs attached to the message for sending replies - /// * `sender_tag` - The anonymous sender tag (for logging/association) - /// * `current_time_ms` - Current time in milliseconds for KCP timing - /// - /// # Returns - /// A tuple containing: - /// - The conv_id extracted from the KCP packet - /// - A vector of decoded KCP packets (for inspection/logging) - /// - A vector of complete reassembled messages ready for IPR processing - pub fn process_incoming( - &mut self, - kcp_data: &[u8], - reply_surbs: Vec, - sender_tag: Option, - current_time_ms: u64, - ) -> Result { - if kcp_data.len() < 4 { - return Err(IpPacketRouterError::KcpError( - "KCP data too short to contain conv_id".to_string(), - )); - } - - // Extract conv_id from first 4 bytes of KCP packet - let conv_id = u32::from_le_bytes([kcp_data[0], kcp_data[1], kcp_data[2], kcp_data[3]]); - - // Get or create session - self.ensure_session(conv_id, sender_tag)?; - - let session = self - .sessions - .get_mut(&conv_id) - .ok_or_else(|| IpPacketRouterError::KcpError("Session not found".to_string()))?; - - session.touch(); - - // Store SURBs for later replies - session.add_surbs(reply_surbs); - - // Input the KCP data and get decoded packets - let decoded_packets = match session.driver.input(kcp_data) { - Ok(pkts) => pkts, - Err(e) => { - log::warn!("KCP input error for conv_id {}: {}", conv_id, e); - return Err(IpPacketRouterError::KcpError(e.to_string())); - } - }; - - // Update KCP state machine - session.driver.update(current_time_ms); - - // Collect any complete messages - let incoming_messages: Vec> = session - .driver - .fetch_incoming() - .into_iter() - .map(|buf| buf.to_vec()) - .collect(); - - Ok(KcpProcessingResult { - conversation_id: conv_id, - decoded_packets, - reassembled_messages: incoming_messages, - }) - } - - /// Wrap outgoing data in KCP for sending via SURB. - /// - /// # Arguments - /// * `conv_id` - The conversation ID - /// * `data` - The data to wrap in KCP - /// * `current_time_ms` - Current time in milliseconds for KCP timing - /// - /// # Returns - /// KCP-encoded packets ready to send - pub fn wrap_response( - &mut self, - conv_id: u32, - data: &[u8], - current_time_ms: u64, - ) -> Result, IpPacketRouterError> { - let session = self - .sessions - .get_mut(&conv_id) - .ok_or_else(|| IpPacketRouterError::KcpError("No session for conv_id".to_string()))?; - - session.touch(); - - // Queue the data for sending - session.driver.send(data); - - // Update to generate outgoing packets - session.driver.update(current_time_ms); - - // Fetch outgoing KCP packets and encode - let packets = session.driver.fetch_outgoing(); - let mut buf = BytesMut::new(); - for pkt in packets { - pkt.encode(&mut buf); - } - - Ok(buf.to_vec()) - } - - /// Take a SURB for sending a reply to a client. - /// - /// # Arguments - /// * `conv_id` - The conversation ID - /// - /// # Returns - /// A SURB if available, None otherwise - pub fn take_surb(&mut self, conv_id: u32) -> Option { - self.sessions.get_mut(&conv_id)?.take_surb() - } - - /// Get the number of available SURBs for a session - pub fn surb_count(&self, conv_id: u32) -> usize { - self.sessions - .get(&conv_id) - .map(|s| s.surb_count()) - .unwrap_or(0) - } - - /// Get the sender_tag associated with a session. - /// - /// Returns None if the session doesn't exist or has no sender_tag. - pub fn get_sender_tag(&self, conv_id: u32) -> Option { - self.sessions.get(&conv_id)?.sender_tag - } - - /// Fetch any pending outgoing KCP packets for a specific session. - /// - /// This is used to send immediate ACKs after receiving packets, - /// rather than waiting for the periodic tick. - pub fn fetch_outgoing_for_conv( - &mut self, - conv_id: u32, - current_time_ms: u64, - ) -> Option> { - let session = self.sessions.get_mut(&conv_id)?; - session.driver.update(current_time_ms); - let packets = session.driver.fetch_outgoing(); - - if packets.is_empty() { - return None; - } - - let mut buf = BytesMut::new(); - for pkt in packets { - pkt.encode(&mut buf); - } - Some(buf.to_vec()) - } - - /// Periodic update for all sessions. - /// - /// This should be called periodically (e.g., every 10-100ms) to: - /// - Drive KCP state machines (retransmissions, etc.) - /// - Clean up expired sessions - /// - /// Returns a list of (conv_id, outgoing_data) pairs for any sessions - /// that have pending outgoing packets. - pub fn tick(&mut self, current_time_ms: u64) -> Vec<(u32, Vec)> { - let mut outgoing = Vec::new(); - - for (&conv_id, session) in self.sessions.iter_mut() { - session.driver.update(current_time_ms); - let packets = session.driver.fetch_outgoing(); - - if !packets.is_empty() { - let mut buf = BytesMut::new(); - for pkt in packets { - pkt.encode(&mut buf); - } - outgoing.push((conv_id, buf.to_vec())); - } - } - - // Clean up expired sessions - self.cleanup_expired(); - - outgoing - } - - /// Remove expired sessions. - pub fn cleanup_expired(&mut self) { - let timeout = self.timeout; - self.sessions.retain(|conv_id, state| { - let expired = state.is_expired(timeout); - if expired { - log::debug!("Removing expired KCP session for conv_id {}", conv_id); - } - !expired - }); - } - - /// Get the number of active sessions - pub fn session_count(&self) -> usize { - self.sessions.len() - } - - /// Check if a session exists for the given conv_id - pub fn has_session(&self, conv_id: u32) -> bool { - self.sessions.contains_key(&conv_id) - } - - /// Ensure a session exists for the given conv_id, creating one if needed - fn ensure_session( - &mut self, - conv_id: u32, - sender_tag: Option, - ) -> Result<(), IpPacketRouterError> { - if self.sessions.contains_key(&conv_id) { - // Update sender_tag if provided - if let Some(tag) = sender_tag - && let Some(session) = self.sessions.get_mut(&conv_id) - { - session.sender_tag = Some(tag); - } - - return Ok(()); - } - - // Check session limit - if self.sessions.len() >= self.max_sessions { - // Try to clean up expired sessions first - self.cleanup_expired(); - - // Still at limit? - if self.sessions.len() >= self.max_sessions { - return Err(IpPacketRouterError::KcpError( - "Maximum KCP sessions reached".to_string(), - )); - } - } - - log::debug!("Creating new KCP session for conv_id {}", conv_id); - let mut state = KcpSessionState::new(conv_id); - state.sender_tag = sender_tag; - self.sessions.insert(conv_id, state); - - Ok(()) - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn test_session_creation() { - let mut manager = KcpSessionManager::new(); - - assert!(!manager.has_session(12345)); - assert_eq!(manager.session_count(), 0); - - // Create a minimal KCP packet (just conv_id) - let conv_id: u32 = 12345; - let mut kcp_data = conv_id.to_le_bytes().to_vec(); - // Add minimal header padding to make it look like a packet - kcp_data.extend_from_slice(&[0u8; 21]); // KCP header is 25 bytes total - - // Processing data should create a session - let result = manager.process_incoming(&kcp_data, vec![], None, 0); - // May error due to invalid KCP packet, but session should be created - let _ = result; - - assert!(manager.has_session(conv_id)); - assert_eq!(manager.session_count(), 1); - } - - #[test] - fn test_session_expiry() { - let mut manager = KcpSessionManager::with_config(Duration::from_millis(10), 100); - let conv_id: u32 = 99999; - - // Create session directly - manager.ensure_session(conv_id, None).unwrap(); - assert!(manager.has_session(conv_id)); - - // Wait for expiry - std::thread::sleep(Duration::from_millis(20)); - - manager.cleanup_expired(); - assert!(!manager.has_session(conv_id)); - } - - #[test] - fn test_max_sessions_limit() { - let mut manager = KcpSessionManager::with_config(Duration::from_secs(300), 2); - - manager.ensure_session(1, None).unwrap(); - manager.ensure_session(2, None).unwrap(); - - assert_eq!(manager.session_count(), 2); - - // Third session should fail - let result = manager.ensure_session(3, None); - assert!(result.is_err()); - } - - #[test] - fn test_kcp_roundtrip() { - use nym_kcp::driver::KcpDriver; - use nym_kcp::session::KcpSession; - - let mut manager = KcpSessionManager::new(); - let conv_id: u32 = 42424242; - - // Create a "client" KCP session to send data - let client_session = KcpSession::new(conv_id); - let mut client_driver = KcpDriver::new(client_session); - - // Client sends a message - let message = b"Hello, IPR via KCP!"; - client_driver.send(message); - client_driver.update(100); - - // Get the KCP packets from the client - let outgoing = client_driver.fetch_outgoing(); - assert!(!outgoing.is_empty(), "Client should produce KCP packets"); - - // Encode packets - let mut kcp_data = BytesMut::new(); - for pkt in outgoing { - pkt.encode(&mut kcp_data); - } - - // Feed to the session manager - let res = manager - .process_incoming(&kcp_data, vec![], None, 100) - .expect("process_incoming should succeed"); - - // Verify conv_id was extracted correctly - assert_eq!(res.conversation_id, conv_id); - - // Should have received the complete message - assert_eq!(res.reassembled_messages.len(), 1); - assert_eq!(res.reassembled_messages[0], message); - } - - #[test] - fn test_surb_storage() { - let mut manager = KcpSessionManager::new(); - let conv_id: u32 = 11111; - - // Create session - manager.ensure_session(conv_id, None).unwrap(); - - // Initially no SURBs - assert_eq!(manager.surb_count(conv_id), 0); - assert!(manager.take_surb(conv_id).is_none()); - - // Note: We can't easily create ReplySurbs in tests without complex setup, - // but the storage mechanism is tested via the session state - } -} diff --git a/service-providers/ip-packet-router/src/lib.rs b/service-providers/ip-packet-router/src/lib.rs index 7810b07e74..fe66577b22 100644 --- a/service-providers/ip-packet-router/src/lib.rs +++ b/service-providers/ip-packet-router/src/lib.rs @@ -17,7 +17,6 @@ pub(crate) mod non_linux_dummy; mod clients; mod constants; mod ip_packet_router; -mod kcp_session_manager; mod mixnet_client; mod mixnet_listener; mod tun_listener; diff --git a/service-providers/ip-packet-router/src/messages/mod.rs b/service-providers/ip-packet-router/src/messages/mod.rs index a695fc5e2f..392e590186 100644 --- a/service-providers/ip-packet-router/src/messages/mod.rs +++ b/service-providers/ip-packet-router/src/messages/mod.rs @@ -9,6 +9,7 @@ pub(crate) enum ClientVersion { V6, V7, V8, + V9, } impl ClientVersion { @@ -17,6 +18,7 @@ impl ClientVersion { ClientVersion::V6 => 6, ClientVersion::V7 => 7, ClientVersion::V8 => 8, + ClientVersion::V9 => 9, } } } diff --git a/service-providers/ip-packet-router/src/messages/request.rs b/service-providers/ip-packet-router/src/messages/request.rs index 9f5704fb40..0f8419e5e1 100644 --- a/service-providers/ip-packet-router/src/messages/request.rs +++ b/service-providers/ip-packet-router/src/messages/request.rs @@ -4,6 +4,7 @@ mod v6; mod v7; mod v8; +mod v9; use nym_ip_packet_requests::{ IpPair, v6::request::IpPacketRequest as IpPacketRequestV6, @@ -129,6 +130,16 @@ impl TryFrom<&ReconstructedMessage> for IpPacketRequest { .ok_or(IpPacketRouterError::MissingSenderTag)?; Ok(IpPacketRequest::from((request_v8, sender_tag))) } + 9 => { + let request_v8 = IpPacketRequestV8::from_reconstructed_message(reconstructed) + .map_err( + |source| IpPacketRouterError::FailedToDeserializeTaggedPacket { source }, + )?; + let sender_tag = reconstructed + .sender_tag + .ok_or(IpPacketRouterError::MissingSenderTag)?; + Ok(v9::convert(request_v8, sender_tag)) + } _ => { log::info!("Received packet with invalid version: v{request_version}"); Err(IpPacketRouterError::InvalidPacketVersion(request_version)) @@ -137,6 +148,21 @@ impl TryFrom<&ReconstructedMessage> for IpPacketRequest { } } +impl IpPacketRequest { + pub(crate) fn version(&self) -> ClientVersion { + match self { + IpPacketRequest::Data(r) => r.version, + IpPacketRequest::Control(c) => match c { + ControlRequest::StaticConnect(r) => r.version, + ControlRequest::DynamicConnect(r) => r.version, + ControlRequest::Disconnect(r) => r.version, + ControlRequest::Ping(r) => r.version, + ControlRequest::Health(r) => r.version, + }, + } + } +} + impl fmt::Display for IpPacketRequest { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { match self { diff --git a/service-providers/ip-packet-router/src/messages/request/v9.rs b/service-providers/ip-packet-router/src/messages/request/v9.rs new file mode 100644 index 0000000000..317fd9b458 --- /dev/null +++ b/service-providers/ip-packet-router/src/messages/request/v9.rs @@ -0,0 +1,27 @@ +// Copyright 2025 - Nym Technologies SA +// SPDX-License-Identifier: GPL-3.0-only + +use nym_ip_packet_requests::v8::request::{ + IpPacketRequest as IpPacketRequestV8, IpPacketRequestData as IpPacketRequestDataV8, +}; +use nym_sdk::mixnet::AnonymousSenderTag; + +use super::{ClientVersion, IpPacketRequest}; + +// v9 uses the same wire format as v8, so we reuse the v8 deserialization +// and just tag the result with ClientVersion::V9. +// +// We cannot implement From<(IpPacketRequestV8, AnonymousSenderTag)> again +// because v8.rs already has that impl (same concrete types). +pub(crate) fn convert( + request: IpPacketRequestV8, + sender_tag: AnonymousSenderTag, +) -> IpPacketRequest { + let version = ClientVersion::V9; + match request.data { + IpPacketRequestDataV8::Data(inner) => IpPacketRequest::Data((inner, version).into()), + IpPacketRequestDataV8::Control(inner) => { + IpPacketRequest::Control((*inner, sender_tag, version).into()) + } + } +} diff --git a/service-providers/ip-packet-router/src/messages/response.rs b/service-providers/ip-packet-router/src/messages/response.rs index 56b6680500..4801983f48 100644 --- a/service-providers/ip-packet-router/src/messages/response.rs +++ b/service-providers/ip-packet-router/src/messages/response.rs @@ -130,6 +130,7 @@ impl VersionedResponse { ClientVersion::V6 => IpPacketResponseV6::try_from(self)?.to_bytes(), ClientVersion::V7 => IpPacketResponseV7::try_from(self)?.to_bytes(), ClientVersion::V8 => IpPacketResponseV8::try_from(self)?.to_bytes(), + ClientVersion::V9 => IpPacketResponseV8::try_from(self)?.to_bytes(), } .map_err(|err| IpPacketRouterError::FailedToSerializeResponsePacket { source: err }) } diff --git a/service-providers/ip-packet-router/src/mixnet_listener.rs b/service-providers/ip-packet-router/src/mixnet_listener.rs index fc5b2b20d6..fa25adbf71 100644 --- a/service-providers/ip-packet-router/src/mixnet_listener.rs +++ b/service-providers/ip-packet-router/src/mixnet_listener.rs @@ -6,7 +6,6 @@ use crate::{ config::Config, constants::DISCONNECT_TIMER_INTERVAL, error::{IpPacketRouterError, Result}, - kcp_session_manager::KcpSessionManager, messages::{ ClientVersion, request::{ @@ -24,6 +23,7 @@ use crate::{ }; use futures::StreamExt; use nym_ip_packet_requests::codec::MultiIpPacketCodec; +use nym_lp::packet::frame::{LpFrameHeader, LpFrameKind, SphinxStreamFrameAttributes}; use nym_sdk::mixnet::MixnetMessageSender; use nym_sphinx::receiver::ReconstructedMessage; use nym_task::ShutdownToken; @@ -31,9 +31,6 @@ use std::{net::SocketAddr, time::Duration}; use tokio::io::AsyncWriteExt; use tokio_util::codec::FramedRead; -/// KCP tick interval for session updates (retransmissions, ACKs, cleanup) -const KCP_TICK_INTERVAL: Duration = Duration::from_millis(10); - #[cfg(not(target_os = "linux"))] type TunDevice = crate::non_linux_dummy::DummyDevice; @@ -60,45 +57,6 @@ pub(crate) struct MixnetListener { // The map of connected clients that the mixnet listener keeps track of. It monitors // activity and disconnects clients that have been inactive for too long. pub(crate) connected_clients: ConnectedClients, - - // KCP session manager for LP clients sending KCP-wrapped messages - pub(crate) kcp_session_manager: KcpSessionManager, -} - -/// Check if a message payload appears to be KCP-wrapped. -/// -/// KCP packets have a 25-byte header with the command byte at position 4. -/// Valid KCP commands are: Push(81), Ack(82), Wask(83), Wins(84). -/// -/// This is distinguishable from IPR protocol messages which have: -/// - Version byte at position 0: 6, 7, or 8 -/// - ServiceProviderType at position 1: 0, 1, or 2 (for v8+) -/// -/// We use a two-step heuristic: -/// 1. Exclude messages that look like IPR protocol headers -/// 2. Check if byte 4 contains a valid KCP command (81-84) -/// -/// See: `Protocol::try_from` in service-provider-requests-common for header format. -fn is_kcp_message(data: &[u8]) -> bool { - // Need at least 25 bytes for KCP header - if data.len() < 25 { - return false; - } - - // First, check if this looks like an IPR protocol message. - // IPR messages have: byte 0 = version (6-8), byte 1 = ServiceProviderType (0-2 for v8+) - // See: IpPacketRequest::try_from in messages/request.rs - let version_byte = data[0]; - let service_type_byte = data[1]; - if (6..=8).contains(&version_byte) && service_type_byte <= 2 { - // This matches IPR protocol header pattern - not a KCP message - return false; - } - - // Check KCP command byte at position 4 - let cmd = data[4]; - // Valid KCP commands: Push=81, Ack=82, Wask=83, Wins=84 - (81..=84).contains(&cmd) } // #[cfg(target_os = "linux")] @@ -172,11 +130,19 @@ impl MixnetListener { let decoder = MultiIpPacketCodec::new(); let mut framed_reader = FramedRead::new(data_request.ip_packets.as_ref(), decoder); - while let Some(Ok(packet)) = framed_reader.next().await { - let result = self - .handle_packet(packet.as_bytes(), data_request.version) - .await; - responses.push(result); + while let Some(result) = framed_reader.next().await { + match result { + Ok(packet) => { + let result = self + .handle_packet(packet.as_bytes(), data_request.version) + .await; + responses.push(result); + } + Err(e) => { + log::warn!("Failed to decode bundled IP packet: {e}"); + continue; + } + } } Ok(responses) @@ -187,6 +153,7 @@ impl MixnetListener { async fn on_static_connect_request( &mut self, connect_request: StaticConnectRequest, + stream_id: Option, ) -> PacketHandleResult { log::info!( "Received static connect request from {}", @@ -209,6 +176,31 @@ impl MixnetListener { let is_client_id_taken = self.connected_clients.is_client_connected(&sent_by); let response = match (is_ip_taken, is_client_id_taken) { + (true, true) if stream_id.is_some() => { + // Stream-mode reconnect: tear down the old handler (which has a + // stale stream_id) and create a fresh one for the new stream. + log::info!("Stream-mode client reconnecting, replacing handler"); + self.connected_clients.disconnect_client(&sent_by); + + let (forward_from_tun_tx, close_tx, handle) = ConnectedClientHandler::start( + sent_by.clone(), + buffer_timeout, + version, + self.mixnet_client.split_sender(), + stream_id, + ); + self.connected_clients.connect( + requested_ips, + sent_by.clone(), + forward_from_tun_tx, + close_tx, + handle, + ); + Response::StaticConnect { + request_id, + reply: StaticConnectResponse::Success, + } + } (true, true) => { log::info!("Connecting an already connected client"); if self @@ -233,6 +225,7 @@ impl MixnetListener { buffer_timeout, version, self.mixnet_client.split_sender(), + stream_id, ); // Register the new client in the set of connected clients @@ -274,9 +267,11 @@ impl MixnetListener { fn on_dynamic_connect_request( &mut self, connect_request: DynamicConnectRequest, + stream_id: Option, ) -> PacketHandleResult { log::info!( - "Received dynamic connect request from {}", + "Received v{} dynamic connect request from {}", + connect_request.version.into_u8(), connect_request.sent_by ); @@ -289,15 +284,22 @@ impl MixnetListener { .unwrap_or(nym_ip_packet_requests::codec::BUFFER_TIMEOUT); if let Some(ips) = self.connected_clients.lookup_ip_from_client_id(&reply_to) { - log::debug!("Reconnecting to the previous session"); - return Ok(Some(VersionedResponse { - version, - reply_to, - response: Response::DynamicConnect { - request_id, - reply: DynamicConnectSuccess { ips }.into(), - }, - })); + if stream_id.is_some() { + // Stream-mode reconnect: tear down old handler (stale stream_id) + // and create a fresh one below with the new stream. + log::info!("Stream-mode client reconnecting, replacing handler"); + self.connected_clients.disconnect_client(&reply_to); + } else { + log::debug!("Reconnecting to the previous session"); + return Ok(Some(VersionedResponse { + version, + reply_to, + response: Response::DynamicConnect { + request_id, + reply: DynamicConnectSuccess { ips }.into(), + }, + })); + } } let Some(new_ips) = self.connected_clients.find_new_ip() else { @@ -318,6 +320,7 @@ impl MixnetListener { buffer_timeout, version, self.mixnet_client.split_sender(), + stream_id, ); // Register the new client in the set of connected clients @@ -403,10 +406,14 @@ impl MixnetListener { })) } - async fn on_control_request(&mut self, control_request: ControlRequest) -> PacketHandleResult { + async fn on_control_request( + &mut self, + control_request: ControlRequest, + stream_id: Option, + ) -> PacketHandleResult { match control_request { - ControlRequest::StaticConnect(r) => self.on_static_connect_request(r).await, - ControlRequest::DynamicConnect(r) => self.on_dynamic_connect_request(r), + ControlRequest::StaticConnect(r) => self.on_static_connect_request(r, stream_id).await, + ControlRequest::DynamicConnect(r) => self.on_dynamic_connect_request(r, stream_id), ControlRequest::Disconnect(r) => self.on_disconnect_request(r), ControlRequest::Ping(r) => self.on_ping_request(r), ControlRequest::Health(r) => self.on_health_request(r), @@ -436,127 +443,106 @@ impl MixnetListener { .unwrap_or("missing".to_owned()) ); - // Check if this is a KCP-wrapped message from an LP client - if is_kcp_message(&reconstructed.message) { - return self.on_kcp_message(reconstructed).await; + // Check if this is an LP Stream frame + if reconstructed.message.len() >= LpFrameHeader::SIZE + && let Ok(header) = LpFrameHeader::parse(&reconstructed.message) + && header.kind == LpFrameKind::SphinxStream + { + return self.on_stream_frame(reconstructed, header).await; } // Regular IPR protocol message (websocket clients) - self.on_ipr_message(reconstructed).await + self.on_ipr_message(reconstructed, None).await } - /// Handle KCP-wrapped messages from LP clients. + /// Handle LP Stream-framed messages. /// - /// LP clients send: KCP(IpPacketRequest) - /// We unwrap the KCP layer, reassemble fragments, then process the inner IpPacketRequest. - /// Responses are wrapped in KCP and sent directly via the sender_tag reply mechanism, - /// rather than being returned for standard handling. - async fn on_kcp_message( + /// Parses stream attributes, processes the inner IPR payload, and handles + /// responses inline (wrapped in LP Stream frames). + async fn on_stream_frame( &mut self, reconstructed: ReconstructedMessage, + header: LpFrameHeader, ) -> Result> { - let current_time_ms = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .map(|d| d.as_millis() as u64) - .unwrap_or(0); - - // Process the KCP data through the session manager. - // AIDEV-NOTE: Reply mechanism for LP clients: - // 1. LP clients MUST send SURBs (in RepliableMessage) when connecting/sending data - // 2. The SDK's client-core layer (ReceivedMessagesBuffer + ReplyController) automatically - // extracts SURBs from incoming RepliableMessages and stores them keyed by sender_tag - // 3. When we call InputMessage::new_reply(sender_tag, ...), the SDK looks up stored SURBs - // 4. The vec![] here is for KcpSessionManager's internal SURB storage (currently unused) - // since the SDK handles SURB storage at a lower layer - // 5. If replies fail, check that LP client is sending SURBs in its messages - let processing_result = self - .kcp_session_manager - .process_incoming( - &reconstructed.message, - vec![], // SDK handles SURB extraction/storage automatically - reconstructed.sender_tag, - current_time_ms, - ) - .inspect_err(|e| { - log::warn!("KCP processing error: {e}"); - })?; - - let conv_id = processing_result.conversation_id; - log::debug!( - "KCP conv_id={}: received {} packets, {} complete messages", - processing_result.conversation_id, - processing_result.decoded_packets.len(), - processing_result.reassembled_messages.len() + "Received LP Stream frame ({} bytes)", + reconstructed.message.len() ); - // Process each reassembled message as an IpPacketRequest - for message_data in processing_result.reassembled_messages { - // Create a synthetic ReconstructedMessage for the inner payload - let inner_reconstructed = ReconstructedMessage { - message: message_data, - sender_tag: reconstructed.sender_tag, - }; + let attrs = SphinxStreamFrameAttributes::parse(&header.frame_attributes).map_err(|e| { + IpPacketRouterError::Other(format!("Invalid stream frame attributes: {e}")) + })?; - match self.on_ipr_message(inner_reconstructed).await { - Ok(results) => { - // Handle responses by wrapping in KCP and sending directly - for result in results { - // false positive: this if can't be collapsed due to `response` being moved - // between calls - #[allow(clippy::collapsible_if)] - if let Ok(Some(response)) = result { - if let Err(e) = self - .handle_kcp_response(conv_id, response, current_time_ms) - .await - { - log::warn!( - "Error sending KCP-wrapped response for conv_id={conv_id}: {e}", - ); - } - } + let stream_id = attrs.stream_id; + log::debug!( + "LP Stream: stream_id={stream_id:#018x}, msg_type={:?}, seq={}", + attrs.msg_type, + attrs.sequence_num + ); + + let payload = &reconstructed.message[LpFrameHeader::SIZE..]; + + // Open frames may carry an empty payload (stream handshake). + if payload.is_empty() { + log::info!("LP Stream: new stream opened (stream_id={stream_id:#018x})"); + return Ok(vec![]); + } + + let inner_reconstructed = ReconstructedMessage { + message: payload.to_vec(), + sender_tag: reconstructed.sender_tag, + }; + + match self + .on_ipr_message(inner_reconstructed, Some(stream_id)) + .await + { + Ok(results) => { + for result in results { + let Ok(Some(response)) = result else { continue }; + if let Err(e) = self.handle_stream_response(stream_id, response).await { + log::warn!( + "Error sending LP Stream response for stream_id={stream_id:#018x}: {e}" + ); } } - Err(e) => { - log::warn!("Error processing KCP inner message: {}", e); - // Continue processing other messages - } + } + Err(e) => { + log::warn!("Error processing LP Stream inner message: {e}"); } } - // Return empty - we handled responses directly above Ok(vec![]) } - /// Wrap a response in KCP and send it via the mixnet reply mechanism. + /// Wrap a response in an LP Stream frame and send it via the mixnet. /// - /// This is used for LP clients that communicate via KCP-wrapped messages. - async fn handle_kcp_response( + /// Used for inline control responses (connect handshake, pong, health) + /// that are sent directly from the message handler — outside the + /// `ConnectedClientHandler` which owns the data-path sequence counter. + /// + /// # Sequence numbering + /// + /// These inline responses always use **seq=0**. The data-path counter in + /// `ConnectedClientHandler` starts at 1 and skips 0 on wrap-around, so + /// the two paths never collide. + /// + /// Limitation: if multiple inline responses are sent on the same stream + /// (e.g. connect + later pong), they share seq=0. The client's reorder + /// buffer will see the second as a duplicate and drop it. In practice + /// this is fine because control responses are rare and idempotent, but + /// if it becomes a problem, give inline responses their own counter. + async fn handle_stream_response( &mut self, - conv_id: u32, + stream_id: u64, response: VersionedResponse, - current_time_ms: u64, ) -> Result<()> { let reply_to = response.reply_to.clone(); - - // Serialize the response let response_bytes = response.try_into_bytes()?; - // Wrap in KCP - let kcp_wrapped = - self.kcp_session_manager - .wrap_response(conv_id, &response_bytes, current_time_ms)?; + let wrapped = crate::clients::encode_stream_frame(stream_id, 0, response_bytes); - log::debug!( - "KCP conv_id={}: wrapped {} byte response into {} bytes", - conv_id, - response_bytes.len(), - kcp_wrapped.len() - ); - - // Send via mixnet using the sender_tag reply mechanism - let input_message = - crate::util::create_message::create_input_message(&reply_to, kcp_wrapped); + let input_message = crate::util::create_message::create_input_message(&reply_to, wrapped); self.mixnet_client.send(input_message).await.map_err(|err| { IpPacketRouterError::FailedToSendPacketToMixnet { @@ -566,9 +552,21 @@ impl MixnetListener { } /// Handle regular IPR protocol messages (from websocket clients). + /// + /// `stream_id` is `Some` when processing a payload extracted from an LP + /// Stream frame, so that connect handlers can thread the id to the + /// `ConnectedClientHandler` for LP-wrapping TUN responses. + /// + /// # Version / transport enforcement + /// + /// - LP Stream frames (`stream_id` is `Some`) **must** carry v9+ payloads. + /// - Non-stream messages (`stream_id` is `None`) **must** be v8 or lower. + /// + /// Messages that violate these rules are dropped. async fn on_ipr_message( &mut self, reconstructed: ReconstructedMessage, + stream_id: Option, ) -> Result> { // First deserialize the request let request = match IpPacketRequest::try_from(&reconstructed) { @@ -579,11 +577,27 @@ impl MixnetListener { req => req, }?; + // Enforce version/transport consistency: + // - LP Stream frames must carry v9+ payloads + // - Non-stream messages must be v8 or lower + let version_num = request.version().into_u8(); + + if stream_id.is_some() && version_num < 9 { + log::warn!("LP Stream frame contains v{version_num} payload, expected v9+; dropping",); + return Ok(vec![]); + } + if stream_id.is_none() && version_num >= 9 { + log::warn!("Non-stream message claims v{version_num}, expected v8 or lower; dropping",); + return Ok(vec![]); + } + log::debug!("Received request: {request}"); match request { IpPacketRequest::Data(request) => self.on_data_request(request).await, - IpPacketRequest::Control(request) => Ok(vec![self.on_control_request(request).await]), + IpPacketRequest::Control(request) => { + Ok(vec![self.on_control_request(request, stream_id).await]) + } } } @@ -640,45 +654,8 @@ impl MixnetListener { } } - /// Handle KCP session tick - drives retransmissions, ACKs, and cleanup. - /// - /// Sends any pending outgoing KCP packets (ACKs, retransmissions) via the - /// sender_tag reply mechanism. - async fn handle_kcp_tick(&mut self) { - let current_time_ms = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .map(|d| d.as_millis() as u64) - .unwrap_or(0); - - // Tick all KCP sessions - generates ACKs, retransmissions, etc. - let outgoing = self.kcp_session_manager.tick(current_time_ms); - - // Send any pending outgoing KCP protocol packets - for (conv_id, data) in outgoing { - // Get the sender_tag for this session to reply via mixnet - let Some(sender_tag) = self.kcp_session_manager.get_sender_tag(conv_id) else { - log::warn!( - "KCP tick: conv_id={} has {} bytes but no sender_tag, dropping", - conv_id, - data.len() - ); - continue; - }; - - log::trace!("KCP tick: conv_id={} sending {} bytes", conv_id, data.len()); - - let reply_to = crate::clients::ConnectedClientId::AnonymousSenderTag(sender_tag); - let input_message = crate::util::create_message::create_input_message(&reply_to, data); - - if let Err(e) = self.mixnet_client.send(input_message).await { - log::warn!("KCP tick: failed to send for conv_id={}: {}", conv_id, e); - } - } - } - pub(crate) async fn run(mut self) -> Result<()> { let mut disconnect_timer = tokio::time::interval(DISCONNECT_TIMER_INTERVAL); - let mut kcp_tick_timer = tokio::time::interval(KCP_TICK_INTERVAL); loop { tokio::select! { @@ -690,9 +667,6 @@ impl MixnetListener { _ = disconnect_timer.tick() => { self.handle_disconnect_timer().await; }, - _ = kcp_tick_timer.tick() => { - self.handle_kcp_tick().await; - }, msg = self.mixnet_client.next() => { if let Some(msg) = msg { match self.on_reconstructed_message(msg).await { @@ -719,98 +693,31 @@ pub(crate) type PacketHandleResult = Result>; #[cfg(test)] mod tests { - use super::*; - #[test] - fn test_is_kcp_message_rejects_ipr_protocol() { - // IPR v8 message: version=8, service_provider_type=1 (IpPacketRouter) - // Even if byte 4 happens to be a valid KCP command, we should reject it - let mut ipr_message = vec![0u8; 30]; - ipr_message[0] = 8; // version - ipr_message[1] = 1; // ServiceProviderType::IpPacketRouter - ipr_message[4] = 81; // This would be KCP Push command, but should be ignored + fn test_lp_stream_frame_detected() { + use bytes::BytesMut; + use nym_lp::packet::frame::{ + LpFrameHeader, LpFrameKind, SphinxStreamFrameAttributes, SphinxStreamMsgType, + }; - assert!( - !is_kcp_message(&ipr_message), - "IPR v8 message should not be detected as KCP" - ); + let attrs = SphinxStreamFrameAttributes { + stream_id: 0x1234, + msg_type: SphinxStreamMsgType::Data, + sequence_num: 42, + }; + let frame = nym_lp::packet::frame::LpFrame::new_stream(attrs, vec![8, 1, 0]); // fake IPR payload + let mut buf = BytesMut::new(); + frame.encode(&mut buf); - // IPR v6 message - ipr_message[0] = 6; - ipr_message[1] = 0; // v6 doesn't use service_provider_type but byte could be 0 - assert!( - !is_kcp_message(&ipr_message), - "IPR v6 message should not be detected as KCP" - ); + let header = LpFrameHeader::parse(&buf).unwrap(); + assert_eq!(header.kind, LpFrameKind::SphinxStream); - // IPR v7 message - ipr_message[0] = 7; - ipr_message[1] = 2; // Authenticator type - assert!( - !is_kcp_message(&ipr_message), - "IPR v7 message should not be detected as KCP" - ); - } + let parsed_attrs = SphinxStreamFrameAttributes::parse(&header.frame_attributes).unwrap(); + assert_eq!(parsed_attrs.stream_id, 0x1234); + assert_eq!(parsed_attrs.msg_type, SphinxStreamMsgType::Data); + assert_eq!(parsed_attrs.sequence_num, 42); - #[test] - fn test_is_kcp_message_accepts_kcp() { - // Valid KCP message: conv_id in bytes 0-3, cmd=Push(81) at byte 4 - // First bytes are conv_id (little-endian u32), so they won't look like IPR version - let mut kcp_message = vec![0u8; 30]; - kcp_message[0] = 0x12; // conv_id byte 0 (not 6-8, so not IPR version) - kcp_message[1] = 0x34; // conv_id byte 1 - kcp_message[2] = 0x56; // conv_id byte 2 - kcp_message[3] = 0x78; // conv_id byte 3 - kcp_message[4] = 81; // KCP Push command - - assert!( - is_kcp_message(&kcp_message), - "Valid KCP message should be detected" - ); - - // Test all valid KCP commands - for cmd in [81u8, 82, 83, 84] { - kcp_message[4] = cmd; - assert!( - is_kcp_message(&kcp_message), - "KCP command {} should be accepted", - cmd - ); - } - } - - #[test] - fn test_is_kcp_message_rejects_short_messages() { - // Less than 25 bytes should be rejected - let short_message = vec![0u8; 24]; - assert!( - !is_kcp_message(&short_message), - "Short message should not be detected as KCP" - ); - - let empty_message: Vec = vec![]; - assert!( - !is_kcp_message(&empty_message), - "Empty message should not be detected as KCP" - ); - } - - #[test] - fn test_is_kcp_message_rejects_invalid_kcp_command() { - // Message with invalid KCP command at byte 4 - let mut message = vec![0u8; 30]; - message[0] = 0x12; // Not IPR version - message[4] = 80; // Invalid KCP command (valid are 81-84) - - assert!( - !is_kcp_message(&message), - "Invalid KCP command should be rejected" - ); - - message[4] = 85; // Also invalid - assert!( - !is_kcp_message(&message), - "Invalid KCP command 85 should be rejected" - ); + // Content is everything after the header + assert_eq!(&buf[LpFrameHeader::SIZE..], &[8, 1, 0]); } }