From 5e63a70215f16c20da2b6dab2ebe0f7608d64fd0 Mon Sep 17 00:00:00 2001 From: mfahampshire Date: Mon, 12 Jan 2026 18:51:01 +0000 Subject: [PATCH] Add mixtcp crate for smoltcp-based IP tunneling Introduces mixtcp, a smoltcp-based device that tunnels TCP/IP through the mixnet via IpMixStream and Exit Gateway IPRs. Components: - NymIprDevice: smoltcp::phy::Device impl using channel-based I/O - NymIprBridge: async task bridging the device to IpMixStream - create_device(): helper to set up the complete stack Examples: - cloudflare_ping: ICMP ping through the mixnet - https_client: HTTPS requests through the mixnet - tls: TLS connections through the mixnet --- Cargo.toml | 8 +- mixtcp/Cargo.toml | 30 +++ mixtcp/README.md | 30 +++ mixtcp/examples/cloudflare_ping.rs | 286 ++++++++++++++++++++++++ mixtcp/examples/https_client.rs | 341 +++++++++++++++++++++++++++++ mixtcp/examples/tls.rs | 270 +++++++++++++++++++++++ mixtcp/src/bridge.rs | 120 ++++++++++ mixtcp/src/device.rs | 165 ++++++++++++++ mixtcp/src/error.rs | 52 +++++ mixtcp/src/lib.rs | 49 +++++ 10 files changed, 1350 insertions(+), 1 deletion(-) create mode 100644 mixtcp/Cargo.toml create mode 100644 mixtcp/README.md create mode 100644 mixtcp/examples/cloudflare_ping.rs create mode 100644 mixtcp/examples/https_client.rs create mode 100644 mixtcp/examples/tls.rs create mode 100644 mixtcp/src/bridge.rs create mode 100644 mixtcp/src/device.rs create mode 100644 mixtcp/src/error.rs create mode 100644 mixtcp/src/lib.rs diff --git a/Cargo.toml b/Cargo.toml index 1d1db9f1da..ef5a16a97e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -148,6 +148,9 @@ members = [ "sdk/ffi/shared", "sdk/rust/nym-sdk", "smolmix/core", + "smolmix/dns", + "smolmix/hyper", + "smolmix/tls", "service-providers/common", "service-providers/ip-packet-router", "service-providers/network-requester", @@ -335,7 +338,7 @@ rayon = "1.5.1" regex = "1.10.6" reqwest = { version = "0.13.1", default-features = false } rs_merkle = "1.5.0" -rustls = { version = "0.23.37", default-features = false } +rustls = { version = "0.23.37", default-features = false, features = ["std", "ring"] } schemars = "0.8.22" semver = "1.0.26" serde = "1.0.219" @@ -350,6 +353,9 @@ serde_plain = "1.0.2" sha2 = "0.10.3" si-scale = "0.2.3" smolmix = { version = "0.0.1", path = "smolmix/core" } +smolmix-dns = { version = "0.0.1", path = "smolmix/dns" } +smolmix-hyper = { version = "0.0.1", path = "smolmix/hyper" } +smolmix-tls = { version = "0.0.1", path = "smolmix/tls" } smoltcp = "0.12" snow = "0.9.6" sphinx-packet = "=0.6.0" diff --git a/mixtcp/Cargo.toml b/mixtcp/Cargo.toml new file mode 100644 index 0000000000..222a447a69 --- /dev/null +++ b/mixtcp/Cargo.toml @@ -0,0 +1,30 @@ +[package] +name = "mixtcp" +version = "0.0.1" +edition = "2021" +license.workspace = true + +[dependencies] +smoltcp = { workspace = true, default-features = false, features = [ + "std", + "medium-ip", + "proto-ipv4", + "proto-ipv6", + "socket-tcp", + "socket-icmp", +] } +tokio = { workspace = true } +bytes = { workspace = true } +tracing = { workspace = true } +tracing-subscriber = { workspace = true } +nym-bin-common = { path = "../common/bin-common", features = ["basic_tracing"] } +nym-sdk = { path = "../sdk/rust/nym-sdk" } +nym-ip-packet-requests = { path = "../common/ip-packet-requests" } +thiserror.workspace = true +rustls = { workspace = true } + +[dev-dependencies] +reqwest.workspace = true +dirs.workspace = true +webpki-roots.workspace = true +serde_json.workspace = true diff --git a/mixtcp/README.md b/mixtcp/README.md new file mode 100644 index 0000000000..9a245b2fde --- /dev/null +++ b/mixtcp/README.md @@ -0,0 +1,30 @@ +# MixTCP + +This is an initial proof of concept of a SmolTCP `device` that uses the Mixnet for transport. It relies on the `IpMixStream` module from the Rust SDK to set up a connection with an Exit Gateway's Ip-Packet-Router, meaning that this is the IP that is seen by the receiver of the request. + +This can be used as the basis for building HTTP(S) crates on top of the Mixnet whilst abstracting away the complexities of using the Mixnet for transport. + +More to come in the future. + +`examples/` contains examples for: +- a TLS ping with Cloudflare +- creating a `reqwest`-like HTTPS `GET` request and receiving a response + +## Component Interaction +```sh + create_device() + | + +--------------+---------------+ + | | | + v v v + NymIprDevice NymIprBridge IpPair + | | (10.0.x.x) + | | + +-- channels --+ + | + v + IpMixStream + | + v + Mixnet +``` diff --git a/mixtcp/examples/cloudflare_ping.rs b/mixtcp/examples/cloudflare_ping.rs new file mode 100644 index 0000000000..2c1d79f2b8 --- /dev/null +++ b/mixtcp/examples/cloudflare_ping.rs @@ -0,0 +1,286 @@ +#![allow(clippy::result_large_err)] +use mixtcp::{create_device, MixtcpError}; +use rustls::{pki_types::ServerName, ClientConfig, ClientConnection}; +use std::{ + io::{self, Read, Write}, + sync::Arc, +}; +use tracing::info; + +use nym_sdk::stream_wrapper::{IpMixStream, NetworkEnvironment}; +use smoltcp::{ + iface::{Config, Interface, SocketSet}, + socket::tcp, + time::Instant, + wire::{HardwareAddress, IpAddress, IpCidr, Ipv4Address}, +}; +use std::sync::Once; +use std::time::Duration; + +static INIT: Once = Once::new(); + +pub struct TlsOverTcp { + pub conn: ClientConnection, +} + +impl TlsOverTcp { + pub fn new(domain: &str) -> Result { + let mut root_store = rustls::RootCertStore::empty(); + root_store.extend(webpki_roots::TLS_SERVER_ROOTS.iter().cloned()); + + let config = ClientConfig::builder() + .with_root_certificates(root_store) + .with_no_client_auth(); + + let server_name = ServerName::try_from(domain) + .map_err(|_| MixtcpError::InvalidDnsName)? + .to_owned(); + + let conn = ClientConnection::new(Arc::new(config), server_name) + .map_err(|_| MixtcpError::TlsHandshakeFailed)?; + + Ok(Self { conn }) + } + + /// Move data from TLS connection to TCP socket + pub fn write_tls(&mut self, socket: &mut tcp::Socket) -> Result<(), MixtcpError> { + let mut buf = [0u8; 4096]; + while self.conn.wants_write() { + match self.conn.write_tls(&mut buf.as_mut_slice()) { + Ok(n) if n > 0 => { + socket + .send_slice(&buf[..n]) + .map_err(|_| MixtcpError::TlsHandshakeFailed)?; + } + _ => break, + } + } + Ok(()) + } + + /// Move data from TCP socket to TLS connection + pub fn read_tls(&mut self, socket: &mut tcp::Socket) -> Result<(), MixtcpError> { + if socket.can_recv() { + let _ = socket.recv(|chunk| { + if !chunk.is_empty() { + inspect_tls_packet(chunk); + let _ = self.conn.read_tls(&mut io::Cursor::new(&mut *chunk)); + let _ = self.conn.process_new_packets(); + } + (chunk.len(), ()) + }); + } + Ok(()) + } + + pub fn send(&mut self, data: &[u8], socket: &mut tcp::Socket) -> Result<(), MixtcpError> { + self.conn + .writer() + .write_all(data) + .map_err(|_| MixtcpError::TlsHandshakeFailed)?; + self.write_tls(socket) + } + + pub fn recv(&mut self, socket: &mut tcp::Socket) -> Result, MixtcpError> { + self.read_tls(socket)?; + let mut result = Vec::new(); + let mut buf = vec![0u8; 4096]; + match self.conn.reader().read(&mut buf) { + Ok(n) if n > 0 => result.extend_from_slice(&buf[..n]), + _ => {} + } + Ok(result) + } +} + +fn inspect_tls_packet(data: &[u8]) { + if data.len() < 5 { + return; + } + let content_type = data[0]; + if !(0x14..=0x17).contains(&content_type) { + return; + } + let version = u16::from_be_bytes([data[1], data[2]]); + let length = u16::from_be_bytes([data[3], data[4]]); + info!( + "TLS packet: ContentType={:#04x}, Version={:#06x}, Length={}", + content_type, version, length + ); + if content_type == 0x16 && data.len() > 5 { + let handshake_type = data[5]; + let handshake_types = match handshake_type { + 0x01 => "ClientHello", + 0x02 => "ServerHello", + 0x0b => "Certificate", + 0x0c => "ServerKeyExchange", + 0x0d => "CertificateRequest", + 0x0e => "ServerHelloDone", + 0x0f => "CertificateVerify", + 0x10 => "ClientKeyExchange", + 0x14 => "Finished", + _ => "Unknown", + }; + info!( + "Handshake type: {:#04x} ({}), Length: {}", + handshake_type, handshake_types, length + ); + } +} + +fn init_logging() { + INIT.call_once_force(|state| { + if state.is_poisoned() { + eprintln!("Logger initialization was poisoned, retrying"); + } + if !tracing::dispatcher::has_been_set() { + nym_bin_common::logging::setup_tracing_logger(); + } + }); +} + +#[tokio::main] +async fn main() -> Result<(), Box> { + init_logging(); + + let ipr_stream = IpMixStream::new(NetworkEnvironment::Mainnet).await?; + let (mut device, bridge, allocated_ips) = create_device(ipr_stream).await?; + info!("Allocated IP: {}", allocated_ips.ipv4); + + tokio::spawn(async move { + bridge.run().await.unwrap(); + }); + + let config = Config::new(HardwareAddress::Ip); + let mut iface = Interface::new(config, &mut device, Instant::now()); + iface.update_ip_addrs(|ip_addrs| { + ip_addrs + .push(IpCidr::new(IpAddress::from(allocated_ips.ipv4), 32)) + .unwrap(); + }); + iface + .routes_mut() + .add_default_ipv4_route(Ipv4Address::UNSPECIFIED) + .unwrap(); + + let tcp_rx_buffer = tcp::SocketBuffer::new(vec![0; 16384]); + let tcp_tx_buffer = tcp::SocketBuffer::new(vec![0; 4096]); + let tcp_socket = tcp::Socket::new(tcp_rx_buffer, tcp_tx_buffer); + let mut sockets = SocketSet::new(vec![]); + let tcp_handle = sockets.add(tcp_socket); + + let target_ip = Ipv4Address::new(1, 1, 1, 1); + let target_port = 443; + + let mut timestamp = Instant::from_millis(0); + let start = tokio::time::Instant::now(); + let mut connected = false; + let mut tls = None; + let mut handshake_completed = false; + let mut request_sent = false; + + loop { + if start.elapsed() > Duration::from_secs(60) { + info!("Test timeout after 60 seconds"); + break; + } + + iface.poll(timestamp, &mut device, &mut sockets); + timestamp += smoltcp::time::Duration::from_millis(1); + let socket = sockets.get_mut::(tcp_handle); + + // TCP connection setup + if !connected && !socket.is_open() { + match socket.connect(iface.context(), (target_ip, target_port), 49152) { + Ok(_) => { + info!("TCP connect started"); + connected = true; + } + Err(e) => { + info!("TCP connect failed: {}", e); + break; + } + } + } + + // TLS setup after TCP established + if socket.state() == tcp::State::Established && tls.is_none() { + info!("TCP established - creating TLS connection"); + match TlsOverTcp::new("cloudflare.com") { + Ok(t) => tls = Some(t), + Err(e) => { + info!("TLS create failed: {}", e); + break; + } + } + } + + // TLS handshake and request + if let Some(ref mut tls_conn) = tls { + let _ = tls_conn.read_tls(socket); + let _ = tls_conn.write_tls(socket); + + // Complete handshake + if !tls_conn.conn.is_handshaking() && !handshake_completed { + handshake_completed = true; + info!("TLS handshake completed - ready for HTTPS"); + + // Send simple HTTP request + let request = b"GET /cdn-cgi/trace HTTP/1.1\r\nHost: cloudflare.com\r\nUser-Agent: mixtcp-test/1.0\r\nAccept: */*\r\nConnection: close\r\n\r\n"; + match tls_conn.send(request, socket) { + Ok(_) => { + info!("HTTPS request sent"); + request_sent = true; + } + Err(e) => { + info!("HTTPS send failed: {}", e); + break; + } + } + } + + // Read response after request sent + if request_sent { + let mut response_data = Vec::new(); + let mut buf = vec![0u8; 4096]; + + match tls_conn.conn.reader().read(&mut buf) { + Ok(0) => { + info!("Response complete - connection closed"); + break; + } + Ok(n) if n > 0 => { + response_data.extend_from_slice(&buf[..n]); + info!("Received {} bytes", n); + + if let Ok(response_str) = std::str::from_utf8(&response_data) { + if response_str.contains("\r\n\r\n") { + info!("HTTPS response received!"); + + if let Some(status_end) = response_str.find("\r\n") { + info!("HTTP Status: {}", &response_str[..status_end]); + } + + info!("Full response: {}", response_str); + return Ok(()); + } + } + } + Ok(1_usize..) => { + todo!() + } + Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => { + // Keep polling + } + Err(e) => { + info!("Read error: {}", e); + break; + } + } + } + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + + Err("No HTTP response received".into()) +} diff --git a/mixtcp/examples/https_client.rs b/mixtcp/examples/https_client.rs new file mode 100644 index 0000000000..c49f83f180 --- /dev/null +++ b/mixtcp/examples/https_client.rs @@ -0,0 +1,341 @@ +#![allow(clippy::result_large_err)] +use mixtcp::{create_device, MixtcpError, NymIprDevice}; +use nym_sdk::stream_wrapper::{IpMixStream, NetworkEnvironment}; +use reqwest::StatusCode; +use rustls::{pki_types::ServerName, ClientConfig, ClientConnection}; +use smoltcp::{ + iface::{Config, Interface, SocketSet}, + socket::tcp, + time::Instant, + wire::{HardwareAddress, IpAddress, IpCidr, Ipv4Address}, +}; +use std::sync::Once; +use std::{ + io::{self, Read, Write}, + sync::Arc, + time::Duration, +}; +use tracing::info; + +static INIT: Once = Once::new(); + +pub struct TlsOverTcp { + pub conn: ClientConnection, +} + +impl TlsOverTcp { + pub fn new(domain: &str) -> Result { + let mut root_store = rustls::RootCertStore::empty(); + root_store.extend(webpki_roots::TLS_SERVER_ROOTS.iter().cloned()); + + let config = ClientConfig::builder() + .with_root_certificates(root_store) + .with_no_client_auth(); + + let server_name = ServerName::try_from(domain) + .map_err(|_| MixtcpError::InvalidDnsName)? + .to_owned(); + + let conn = ClientConnection::new(Arc::new(config), server_name) + .map_err(|_| MixtcpError::TlsHandshakeFailed)?; + + Ok(Self { conn }) + } + + pub fn write_tls(&mut self, socket: &mut tcp::Socket) -> Result<(), MixtcpError> { + let mut buf = [0u8; 4096]; + while self.conn.wants_write() { + match self.conn.write_tls(&mut buf.as_mut_slice()) { + Ok(n) if n > 0 => { + socket + .send_slice(&buf[..n]) + .map_err(|_| MixtcpError::TlsHandshakeFailed)?; + } + _ => break, + } + } + Ok(()) + } + + pub fn read_tls(&mut self, socket: &mut tcp::Socket) -> Result<(), MixtcpError> { + if socket.can_recv() { + let _ = socket.recv(|chunk| { + if !chunk.is_empty() { + let _ = self.conn.read_tls(&mut io::Cursor::new(&mut *chunk)); + let _ = self.conn.process_new_packets(); + } + (chunk.len(), ()) + }); + } + Ok(()) + } + + pub fn send(&mut self, data: &[u8], socket: &mut tcp::Socket) -> Result<(), MixtcpError> { + self.conn + .writer() + .write_all(data) + .map_err(|_| MixtcpError::TlsHandshakeFailed)?; + self.write_tls(socket) + } +} + +/// Reqwest-ish client right now, just a handrolled GET request for the example +pub struct MixtcpReqwestClient { + device: Arc>, + _bridge: tokio::task::JoinHandle<()>, + _allocated_ip: Ipv4Address, +} + +impl MixtcpReqwestClient { + pub async fn new() -> Result { + let ipr_stream = IpMixStream::new(NetworkEnvironment::Mainnet) + .await + .map_err(|_| MixtcpError::MixnetConnectionFailed)?; + + let (mut device, bridge, allocated_ips) = create_device(ipr_stream).await?; + info!("Allocated IP: {}", allocated_ips.ipv4); + + let bridge_handle = tokio::spawn(async move { + if let Err(e) = bridge.run().await { + tracing::error!("Bridge error: {}", e); + } + }); + + let config = Config::new(HardwareAddress::Ip); + let mut iface = Interface::new(config, &mut device, Instant::now()); + iface.update_ip_addrs(|ip_addrs| { + ip_addrs + .push(IpCidr::new(IpAddress::from(allocated_ips.ipv4), 32)) + .unwrap(); + }); + iface + .routes_mut() + .add_default_ipv4_route(Ipv4Address::UNSPECIFIED) + .unwrap(); + + let device = Arc::new(tokio::sync::Mutex::new((iface, device))); + + Ok(Self { + device, + _bridge: bridge_handle, + _allocated_ip: allocated_ips.ipv4, + }) + } + + pub async fn get(&self, url: &str) -> Result { + let parsed_url = reqwest::Url::parse(url).map_err(|_| MixtcpError::InvalidUrl)?; + let host = parsed_url.host_str().ok_or(MixtcpError::InvalidUrl)?; + let path = parsed_url.path(); + + let response_bytes = self.simple_get_request(host, path).await?; + let (status, body) = self.parse_simple_response(&response_bytes)?; + + Ok(MixtcpResponse { status, body }) + } + + async fn simple_get_request(&self, domain: &str, path: &str) -> Result, MixtcpError> { + let tcp_rx_buffer = tcp::SocketBuffer::new(vec![0; 16384]); + let tcp_tx_buffer = tcp::SocketBuffer::new(vec![0; 4096]); + let tcp_socket = tcp::Socket::new(tcp_rx_buffer, tcp_tx_buffer); + let mut sockets = SocketSet::new(vec![]); + let tcp_handle = sockets.add(tcp_socket); + + let target_ip = Ipv4Address::new(1, 1, 1, 1); + let target_port = 443; + + let mut timestamp = Instant::from_millis(0); + let start = tokio::time::Instant::now(); + let mut connected = false; + let mut tls = None; + let mut handshake_completed = false; + let mut request_sent = false; + let mut response_data = Vec::new(); + + let mut device_guard = self.device.lock().await; + let (ref mut iface, ref mut device) = &mut *device_guard; + + loop { + if start.elapsed() > Duration::from_secs(60) { + return Err(MixtcpError::Timeout); + } + + iface.poll(timestamp, device, &mut sockets); + timestamp += smoltcp::time::Duration::from_millis(1); + let socket = sockets.get_mut::(tcp_handle); + + if !connected && !socket.is_open() { + match socket.connect(iface.context(), (target_ip, target_port), 49152) { + Ok(_) => { + info!("TCP connect started"); + connected = true; + } + Err(e) => { + info!("TCP connect failed: {}", e); + return Err(MixtcpError::TcpConnectionFailed); + } + } + } + + if socket.state() == tcp::State::Established && tls.is_none() { + info!("TCP established - creating TLS connection"); + match TlsOverTcp::new(domain) { + Ok(t) => tls = Some(t), + Err(e) => { + info!("TLS create failed: {}", e); + return Err(MixtcpError::TlsHandshakeFailed); + } + } + } + + if let Some(ref mut tls_conn) = tls { + let _ = tls_conn.read_tls(socket); + let _ = tls_conn.write_tls(socket); + + if !tls_conn.conn.is_handshaking() && !handshake_completed { + handshake_completed = true; + info!("TLS handshake completed - ready for HTTPS"); + + let request = format!( + "GET {} HTTP/1.1\r\nHost: {}\r\nUser-Agent: mixtcp/1.0\r\nAccept: */*\r\nConnection: close\r\n\r\n", + path, domain + ); + tls_conn.send(request.as_bytes(), socket)?; + info!("HTTPS request sent"); + request_sent = true; + } + + if request_sent { + let mut buf = vec![0u8; 4096]; + match tls_conn.conn.reader().read(&mut buf) { + Ok(0) => { + info!("Response complete"); + break; + } + Ok(n) if n > 0 => { + response_data.extend_from_slice(&buf[..n]); + if let Ok(response_str) = std::str::from_utf8(&response_data) { + if response_str.contains("\r\n\r\n") { + return Ok(response_data); + } + } + } + Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => {} + Err(e) => { + info!("Read error: {}", e); + return Err(MixtcpError::ResponseReadFailed); + } + Ok(_) => continue, + } + } + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + + Err(MixtcpError::NoResponseReceived) + } + + /// Simple response - just extract status and body + fn parse_simple_response(&self, response_bytes: &[u8]) -> Result<(u16, String), MixtcpError> { + let response_str = String::from_utf8_lossy(response_bytes); + + let status_line = response_str + .lines() + .next() + .ok_or(MixtcpError::InvalidHttpResponse)?; + + let status: u16 = status_line + .split_whitespace() + .nth(1) + .and_then(|s| s.parse().ok()) + .unwrap_or(200); + + if let Some(body_start) = response_str.find("\r\n\r\n") { + let body = response_str[body_start + 4..].to_string(); + Ok((status, body)) + } else { + Err(MixtcpError::InvalidHttpResponse) + } + } +} + +pub struct MixtcpResponse { + status: u16, + body: String, +} + +impl MixtcpResponse { + pub fn status(&self) -> StatusCode { + StatusCode::from_u16(self.status).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR) + } + + pub async fn text(self) -> Result { + Ok(self.body) + } +} + +fn init_logging() { + INIT.call_once_force(|state| { + if state.is_poisoned() { + eprintln!("Logger initialization was poisoned, retrying"); + } + if !tracing::dispatcher::has_been_set() { + nym_bin_common::logging::setup_tracing_logger(); + } + }); +} + +#[tokio::main] +async fn main() -> Result<(), Box> { + init_logging(); + + let test_url = "https://cloudflare.com/cdn-cgi/trace"; + + info!("Fetching with plain reqwest..."); + let start = tokio::time::Instant::now(); + let plain_response = reqwest::get(test_url).await?; + let plain_status = plain_response.status(); + let plain_text = plain_response.text().await?; + let plain_duration = start.elapsed(); + + info!( + "Plain reqwest - Status: {}, Time: {:?}", + plain_status, plain_duration + ); + + info!("Setting up mixnet client..."); + let client = MixtcpReqwestClient::new().await?; + let start = tokio::time::Instant::now(); + let mixnet_response = client.get(test_url).await?; + let mixnet_status = mixnet_response.status(); + let mixnet_text = mixnet_response.text().await?; + let mixnet_duration = start.elapsed(); + + info!( + "Mixnet reqwest - Status: {}, Time: {:?}", + mixnet_status, mixnet_duration + ); + + info!("Status codes match: {}", plain_status == mixnet_status); + info!( + "Response lengths match: {}", + plain_text.len() == mixnet_text.len() + ); + + let key_fields = ["fl=", "ip=", "ts=", "visit_scheme="]; + for field in key_fields { + let plain_has = plain_text.contains(field); + let mixnet_has = mixnet_text.contains(field); + info!( + "Field '{}' - Plain: {}, Mixnet: {}", + field, plain_has, mixnet_has + ); + assert_eq!(plain_has, mixnet_has, "Field '{}' mismatch", field); + } + + info!("Plain reqwest time: {:?}", plain_duration); + info!("Mixnet reqwest time: {:?}", mixnet_duration); + let slowdown = mixnet_duration.as_millis() as f64 / plain_duration.as_millis() as f64; + info!("Mixnet slowdown: {:.1}x", slowdown); + info!("Both responses match"); + Ok(()) +} diff --git a/mixtcp/examples/tls.rs b/mixtcp/examples/tls.rs new file mode 100644 index 0000000000..749e9ecc14 --- /dev/null +++ b/mixtcp/examples/tls.rs @@ -0,0 +1,270 @@ +#![allow(clippy::result_large_err)] +use mixtcp::{create_device, MixtcpError}; +use rustls::{pki_types::ServerName, ClientConfig, ClientConnection}; +use std::{ + io::{self, Read, Write}, + sync::Arc, +}; +use tracing::info; + +use nym_sdk::stream_wrapper::{IpMixStream, NetworkEnvironment}; + +use smoltcp::{ + iface::{Config, Interface, SocketSet}, + socket::tcp, + time::Instant, + wire::{HardwareAddress, IpAddress, IpCidr, Ipv4Address}, +}; +use std::sync::Once; +use std::time::Duration; + +static INIT: Once = Once::new(); + +pub struct TlsOverTcp { + pub conn: ClientConnection, +} + +impl TlsOverTcp { + pub fn new(domain: &str) -> Result { + let mut root_store = rustls::RootCertStore::empty(); + root_store.extend(webpki_roots::TLS_SERVER_ROOTS.iter().cloned()); + + let config = ClientConfig::builder() + .with_root_certificates(root_store) + .with_no_client_auth(); + + let server_name = ServerName::try_from(domain) + .map_err(|_| MixtcpError::InvalidDnsName)? + .to_owned(); + + let conn = ClientConnection::new(Arc::new(config), server_name) + .map_err(|_| MixtcpError::TlsHandshakeFailed)?; + + Ok(Self { conn }) + } + + /// Move data from TLS connection to TCP socket + pub fn write_tls(&mut self, socket: &mut tcp::Socket) -> Result<(), MixtcpError> { + let mut buf = [0u8; 4096]; + while self.conn.wants_write() { + match self.conn.write_tls(&mut buf.as_mut_slice()) { + Ok(n) if n > 0 => { + socket + .send_slice(&buf[..n]) + .map_err(|_| MixtcpError::TlsHandshakeFailed)?; + } + _ => break, + } + } + Ok(()) + } + + /// Move data from TCP socket to TLS connection + pub fn read_tls(&mut self, socket: &mut tcp::Socket) -> Result<(), MixtcpError> { + if socket.can_recv() { + let _ = socket.recv(|chunk| { + if !chunk.is_empty() { + inspect_tls_packet(chunk); + let _ = self.conn.read_tls(&mut io::Cursor::new(&mut *chunk)); + let _ = self.conn.process_new_packets(); + } + (chunk.len(), ()) + }); + } + Ok(()) + } + + pub fn send(&mut self, data: &[u8], socket: &mut tcp::Socket) -> Result<(), MixtcpError> { + self.conn + .writer() + .write_all(data) + .map_err(|_| MixtcpError::TlsHandshakeFailed)?; + self.write_tls(socket) + } + + pub fn recv(&mut self, socket: &mut tcp::Socket) -> Result, MixtcpError> { + self.read_tls(socket)?; + let mut result = Vec::new(); + let mut buf = vec![0u8; 4096]; + match self.conn.reader().read(&mut buf) { + Ok(n) if n > 0 => result.extend_from_slice(&buf[..n]), + _ => {} + } + Ok(result) + } +} + +fn inspect_tls_packet(data: &[u8]) { + if data.len() < 5 { + return; + } + let content_type = data[0]; + if !(0x14..=0x17).contains(&content_type) { + return; + } + let version = u16::from_be_bytes([data[1], data[2]]); + let length = u16::from_be_bytes([data[3], data[4]]); + info!( + "TLS packet: ContentType={:#04x}, Version={:#06x}, Length={}", + content_type, version, length + ); + if content_type == 0x16 && data.len() > 5 { + let handshake_type = data[5]; + let handshake_types = match handshake_type { + 0x01 => "ClientHello", + 0x02 => "ServerHello", + 0x0b => "Certificate", + 0x0c => "ServerKeyExchange", + 0x0d => "CertificateRequest", + 0x0e => "ServerHelloDone", + 0x0f => "CertificateVerify", + 0x10 => "ClientKeyExchange", + 0x14 => "Finished", + _ => "Unknown", + }; + info!( + "Handshake type: {:#04x} ({}), Length: {}", + handshake_type, handshake_types, length + ); + } +} + +fn init_logging() { + INIT.call_once_force(|state| { + if state.is_poisoned() { + eprintln!("Logger initialization was poisoned, retrying"); + } + if !tracing::dispatcher::has_been_set() { + nym_bin_common::logging::setup_tracing_logger(); + } + }); +} + +#[tokio::main] +async fn main() -> Result<(), Box> { + init_logging(); + + let ipr_stream = IpMixStream::new(NetworkEnvironment::Mainnet).await?; + let (mut device, bridge, allocated_ips) = create_device(ipr_stream).await?; + info!("Allocated IP: {}", allocated_ips.ipv4); + + tokio::spawn(async move { + bridge.run().await.unwrap(); + }); + + let config = Config::new(HardwareAddress::Ip); + let mut iface = Interface::new(config, &mut device, Instant::now()); + iface.update_ip_addrs(|ip_addrs| { + ip_addrs + .push(IpCidr::new(IpAddress::from(allocated_ips.ipv4), 32)) + .unwrap(); + }); + iface + .routes_mut() + .add_default_ipv4_route(Ipv4Address::UNSPECIFIED) + .unwrap(); + + let tcp_rx_buffer = tcp::SocketBuffer::new(vec![0; 16384]); + let tcp_tx_buffer = tcp::SocketBuffer::new(vec![0; 4096]); + let tcp_socket = tcp::Socket::new(tcp_rx_buffer, tcp_tx_buffer); + let mut sockets = SocketSet::new(vec![]); + let tcp_handle = sockets.add(tcp_socket); + + let target_ip = Ipv4Address::new(1, 1, 1, 1); // Pinging Cloudflare + let target_port = 443; + info!("Connecting to {}:{} through mixnet", target_ip, target_port); + + let mut timestamp = Instant::from_millis(0); + let start = tokio::time::Instant::now(); + let mut connected = false; + let mut tls = None; + let handshake_completed = false; + + loop { + if start.elapsed() > Duration::from_secs(120) { + info!("Test timeout after 120 seconds"); + break; + } + + iface.poll(timestamp, &mut device, &mut sockets); + timestamp += smoltcp::time::Duration::from_millis(1); + let socket = sockets.get_mut::(tcp_handle); + + if !connected && !socket.is_open() { + match socket.connect(iface.context(), (target_ip, target_port), 49152) { + Ok(_) => { + info!("TCP connect started"); + connected = true; + } + Err(e) => { + info!("TCP connect failed: {}", e); + break; + } + } + } + + if start.elapsed().as_secs().is_multiple_of(5) && start.elapsed().as_millis() % 1000 < 100 { + info!( + "State: TCP={:?}, established={}, can_send={}, can_recv={}", + socket.state(), + socket.state() == tcp::State::Established, + socket.may_send(), + socket.can_recv() + ); + } + + if socket.state() == tcp::State::Established && tls.is_none() { + info!("TCP established - creating TLS connection"); + match TlsOverTcp::new("cloudflare.com") { + Ok(t) => tls = Some(t), + Err(e) => { + info!("TLS create failed: {}", e); + break; + } + } + } + + if let Some(ref mut tls_conn) = tls { + let _ = tls_conn.read_tls(socket); + let _ = tls_conn.write_tls(socket); + + if start.elapsed().as_secs().is_multiple_of(10) + && start.elapsed().as_millis() % 1000 < 100 + { + info!( + "TLS state: handshaking={}, wants_read={}, wants_write={}", + tls_conn.conn.is_handshaking(), + tls_conn.conn.wants_read(), + tls_conn.conn.wants_write() + ); + } + + if !tls_conn.conn.is_handshaking() && !handshake_completed { + info!("TLS handshake complete"); + info!( + "TLS verification: handshake_complete=true, wants_read={}, wants_write={}", + tls_conn.conn.wants_read(), + tls_conn.conn.wants_write() + ); + + match tls_conn.recv(socket) { + Ok(data) if data.is_empty() => { + info!("No unexpected application data waiting to be read"); + } + Ok(data) => { + info!("Unexpected application data received: {} bytes", data.len()); + } + Err(e) => { + info!("TLS recv check failed: {}", e); + } + } + info!("TLS handshake successful with cloudflare"); + break; + } + } + tokio::time::sleep(Duration::from_millis(1)).await; + } + + info!("Test completed"); + Ok(()) +} diff --git a/mixtcp/src/bridge.rs b/mixtcp/src/bridge.rs new file mode 100644 index 0000000000..aea60167b4 --- /dev/null +++ b/mixtcp/src/bridge.rs @@ -0,0 +1,120 @@ +// Copyright 2024 - Nym Technologies SA +// SPDX-License-Identifier: GPL-2.0-only + +use crate::error::MixtcpError; +use nym_ip_packet_requests::codec::MultiIpPacketCodec; +use nym_sdk::stream_wrapper::IpMixStream; +use tokio::sync::mpsc; +use tracing::{error, info}; + +/// Asynchronous bridge between smoltcp device and Mixnet. +/// +/// This component runs in a separate task and handles all asynchronous +/// operations required for outbound communication. It receives packets +/// from the device via channels, bundles them according to IPR protocol +/// (MultiIpPacketCodec) and transmits them through the Mixnet. +/// +/// # Packet Processing Flow +/// +/// Outgoing packets: +/// - Receive from device via channel +/// - Bundle using MultiIpPacketCodec +/// - Send through mixnet via send_ip_packet() +/// +/// Incoming packets: +/// - Poll mixnet with handle_incoming() +/// - Forward to device via channel +/// - Device queues for smoltcp consumption +pub struct NymIprBridge { + /// Connected IPR stream for mixnet communication + stream: IpMixStream, + /// Channel for receiving outgoing packets from device + tx_receiver: mpsc::UnboundedReceiver>, + /// Channel for sending incoming packets to device + rx_sender: mpsc::UnboundedSender>, +} + +impl NymIprBridge { + pub fn new( + stream: IpMixStream, + tx_receiver: mpsc::UnboundedReceiver>, + rx_sender: mpsc::UnboundedSender>, + ) -> Self { + Self { + stream, + tx_receiver, + rx_sender, + } + } + + /// Runs the bridge event loop. + /// + /// This method should be spawned in a separate task. It continuously: + /// - Processes outgoing packets from the device + /// - Polls for incoming packets from the mixnet + /// - Maintains packet statistics + /// + /// The loop exits when channels are closed or an error occurs. + pub async fn run(mut self) -> Result<(), MixtcpError> { + info!("Starting Nym IPR bridge"); + let mut packets_sent = 0; + let mut packets_received = 0; + + loop { + tokio::select! { + // Outgoing packets from smoltcp layer above. + Some(packet) = self.tx_receiver.recv() => { + info!("Bridge sending {} byte packet to mixnet", packet.len()); + + // Log packet details for debugging + if packet.len() >= 20 { + let version = (packet[0] >> 4) & 0xF; + let proto = packet[9]; + let src_ip = &packet[12..16]; + let dst_ip = &packet[16..20]; + info!( + "Outgoing IPv{} packet: proto={}, src={}.{}.{}.{}, dst={}.{}.{}.{}", + version, proto, + src_ip[0], src_ip[1], src_ip[2], src_ip[3], + dst_ip[0], dst_ip[1], dst_ip[2], dst_ip[3] + ); + } + + // Necessary to bundle for IPR! See stream_wrapper_ipr.rs tests. + let bundled = MultiIpPacketCodec::bundle_one_packet(packet.into()); + if let Err(e) = self.stream.send_ip_packet(&bundled).await { + error!("Failed to send packet through mixnet: {}", e); + } else { + packets_sent += 1; + info!("Total packets sent: {}", packets_sent); + } + } + + // Poll for incoming packets from mixnet + Ok(packets) = self.stream.handle_incoming() => { + if !packets.is_empty() { + info!("Bridge received {} packets from mixnet", packets.len()); + for packet in packets { + info!("Incoming packet: {} bytes", packet.len()); + + // Forward to device via channel + if self.rx_sender.send(packet.to_vec()).is_err() { + error!("Failed to send packet to device - receiver dropped"); + return Err(MixtcpError::ChannelClosed); + } + packets_received += 1; + info!("Total packets received: {}", packets_received); + } + } + } + + else => { + info!("Bridge shutting down. Sent: {}, Received: {}", packets_sent, packets_received); + break; + } + } + } + + Ok(()) + } +} diff --git a/mixtcp/src/device.rs b/mixtcp/src/device.rs new file mode 100644 index 0000000000..f020459a02 --- /dev/null +++ b/mixtcp/src/device.rs @@ -0,0 +1,165 @@ +// Copyright 2024 - Nym Technologies SA +// SPDX-License-Identifier: GPL-2.0-only + +use smoltcp::{ + phy::{Device, DeviceCapabilities, Medium, RxToken, TxToken}, + time::Instant, +}; +use std::collections::VecDeque; +use tokio::sync::mpsc; +use tracing::{info, warn}; + +/// # Overview +/// We need something to bridge the async / sync weirdness (Device trait fns are sync, IpMixStream fns are +/// async) in a way that allows for the `NymIprDevice` to look and act like any other device. +/// +/// We need to be polling the queue to/from the NymIprBridge, hence the addition of the +/// mpsc channels in the Device struct and the extra fns. +/// +/// # Architecture +/// smoltcp (sync) <-> NymIprDevice <-> channels <-> NymIprBridge <-> Mixnet (async) +/// +/// The device maintains a receive queue for packets coming from the mixnet and +/// uses unbounded channels to communicate with the bridge task that handles the +/// actual mixnet I/O. We poll the channel in receive() to move packets via mpsc +/// from async to sync world. +/// +/// This way no blocking from smoltcp + allows for concurrency. +/// +/// Adapter pattern between sync polling-based I/O and async event-based I/O. +pub struct NymIprDevice { + // Receive queue for packets coming from the mixnet + rx_queue: VecDeque>, + + // Channel to send packets to the bridge task + tx_sender: mpsc::UnboundedSender>, + + // Device capabilities + capabilities: DeviceCapabilities, + + // Channel to receive packets from the bridge task + rx_receiver: mpsc::UnboundedReceiver>, +} + +impl NymIprDevice { + pub fn new( + tx_sender: mpsc::UnboundedSender>, + rx_receiver: mpsc::UnboundedReceiver>, + ) -> Self { + let mut capabilities = DeviceCapabilities::default(); + capabilities.medium = Medium::Ip; + // Standard MTU for IP packets - TODO make configurable + capabilities.max_transmission_unit = 1500; + // Process one packet at a time. TODO experiment with this + capabilities.max_burst_size = Some(1); + + Self { + rx_queue: VecDeque::new(), + tx_sender, + capabilities, + rx_receiver, + } + } + + /// Poll for new packets from the bridge + fn poll_rx_queue(&mut self) { + // Try to receive all available packets without blocking, queue them for smoltcp consumption. + while let Ok(packet) = self.rx_receiver.try_recv() { + info!("Received packet of {} bytes from bridge", packet.len()); + self.rx_queue.push_back(packet); + } + } + + pub fn tx_sender(&self) -> mpsc::UnboundedSender> { + self.tx_sender.clone() + } + + /// Get the receiver for external use + pub fn rx_receiver(&self) -> mpsc::UnboundedReceiver> { + // Create a new channel and return the receiver + // This is a bit of a hack but necessary for the current architecture + let (_tx, rx) = mpsc::unbounded_channel(); + // We just need the receiver for testing + rx + } +} + +impl Device for NymIprDevice { + type RxToken<'a> + = NymRxToken + where + Self: 'a; + type TxToken<'a> + = NymTxToken + where + Self: 'a; + + fn receive(&mut self, _timestamp: Instant) -> Option<(Self::RxToken<'_>, Self::TxToken<'_>)> { + // Poll for new packets from the async bridge + self.poll_rx_queue(); + + // Check if we have a packet to deliver + let packet = self.rx_queue.pop_front()?; + + // Create tokens - RxToken owns the packet data + let rx_token = NymRxToken { buffer: packet }; + let tx_token = NymTxToken { + tx_sender: self.tx_sender.clone(), + }; + + Some((rx_token, tx_token)) + } + + fn transmit(&mut self, _timestamp: Instant) -> Option> { + // We can always transmit (channel will buffer) + Some(NymTxToken { + tx_sender: self.tx_sender.clone(), + }) + } + + fn capabilities(&self) -> DeviceCapabilities { + self.capabilities.clone() + } +} + +/// Receive token - owns the packet buffer +pub struct NymRxToken { + buffer: Vec, +} + +impl RxToken for NymRxToken { + fn consume(self, f: F) -> R + where + F: FnOnce(&[u8]) -> R, + { + info!("Consuming RX packet of {} bytes", self.buffer.len()); + f(&self.buffer) + } +} + +/// Transmit token - holds channel sender +pub struct NymTxToken { + tx_sender: mpsc::UnboundedSender>, +} + +impl TxToken for NymTxToken { + fn consume(self, len: usize, f: F) -> R + where + F: FnOnce(&mut [u8]) -> R, + { + // Create buffer for the packet + let mut buffer = vec![0u8; len]; + + // Let smoltcp fill the packet + let result = f(&mut buffer); + + // Send raw packet to the bridge task for transmission + if let Err(e) = self.tx_sender.send(buffer) { + warn!("Failed to send packet to bridge: {}", e); + } else { + info!("Sent {} byte packet to bridge", len); + } + + result + } +} diff --git a/mixtcp/src/error.rs b/mixtcp/src/error.rs new file mode 100644 index 0000000000..56d80807d1 --- /dev/null +++ b/mixtcp/src/error.rs @@ -0,0 +1,52 @@ +// Copyright 2024 - Nym Technologies SA +// SPDX-License-Identifier: GPL-2.0-only + +use thiserror::Error; + +#[derive(Error, Debug)] +pub enum MixtcpError { + #[error("Channel closed")] + ChannelClosed, + + #[error("Not connected to IPR")] + NotConnected, + + #[error("Nym SDK error: {0}")] + NymSdk(#[from] nym_sdk::Error), + + #[error("TLS handshake failed")] + TlsHandshakeFailed, + + #[error("TLS encrypt/decrypt error")] + TlsCrypto, + + #[error("DNS err placeholder")] + InvalidDnsName, + + #[error("IO error: {0}")] + Io(#[from] std::io::Error), + + #[error("HTTP parse failed")] + HttpParseFailed, + + #[error("Invalid URL")] + InvalidUrl, + + #[error("Mixnet connection failed")] + MixnetConnectionFailed, + + #[error("Request timeout")] + Timeout, + + #[error("TCP connection failed")] + TcpConnectionFailed, + + #[error("Response read failed")] + ResponseReadFailed, + + #[error("No response received")] + NoResponseReceived, + + #[error("Invalid HTTP response")] + InvalidHttpResponse, +} diff --git a/mixtcp/src/lib.rs b/mixtcp/src/lib.rs new file mode 100644 index 0000000000..33fa66cba0 --- /dev/null +++ b/mixtcp/src/lib.rs @@ -0,0 +1,49 @@ +// Copyright 2024 - Nym Technologies SA +// SPDX-License-Identifier: GPL-2.0-only + +mod bridge; +mod device; +mod error; + +pub use bridge::NymIprBridge; +pub use device::NymIprDevice; +pub use error::MixtcpError; + +use nym_ip_packet_requests::IpPair; +use nym_sdk::stream_wrapper::IpMixStream; +use tokio::sync::mpsc; + +/// Create a connected smoltcp device and async bridge for the tunneling packets through the +/// Mixnet to remote hosts via an IPR. +/// +/// This function handles the complete setup process: +/// - Ensures the IPR stream is connected +/// - Retrieves allocated IP addresses +/// - Creates communication channels +/// - Constructs the device and bridge components +pub async fn create_device( + mut ipr_stream: IpMixStream, +) -> Result<(NymIprDevice, NymIprBridge, IpPair), MixtcpError> { + // Ensure the stream is connected + if !ipr_stream.is_connected() { + ipr_stream.connect_tunnel().await?; + } + + // Get the allocated IPs before moving the stream - need these for proper packet creation + // further 'up' the flow in the code calling this fn (see examples/tcp_connect.rs). + let allocated_ips = *ipr_stream + .allocated_ips() + .ok_or(MixtcpError::NotConnected)?; + + // Create channels for device <-> bridge communication + let (tx_to_bridge, tx_from_device) = mpsc::unbounded_channel(); + let (rx_to_device, rx_from_bridge) = mpsc::unbounded_channel(); + + // Create device + let device = NymIprDevice::new(tx_to_bridge, rx_from_bridge); + + // Create bridge (moves ipr_stream) + let bridge = NymIprBridge::new(ipr_stream, tx_from_device, rx_to_device); + + Ok((device, bridge, allocated_ips)) +}