GRIN/nym
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add mixtcp crate for smoltcp-based IP tunneling

Browse Source 
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
This commit is contained in:
mfahampshire
2026-01-12 18:51:01 +00:00
parent 7efb81a601
commit 5e63a70215
10 changed files with 1350 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.toml
+7 -1
View File
@@ -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"
mixtcp/Cargo.toml
+30
View File
@@ -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
mixtcp/README.md
+30
View File
@@ -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
```
mixtcp/examples/cloudflare_ping.rs
+286
View File
@@ -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<Self, MixtcpError> {
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<Vec<u8>, 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<dyn std::error::Error>> {
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::Socket>(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())
}
mixtcp/examples/https_client.rs
+341
View File
@@ -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<Self, MixtcpError> {
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<tokio::sync::Mutex<(smoltcp::iface::Interface, NymIprDevice)>>,
_bridge: tokio::task::JoinHandle<()>,
_allocated_ip: Ipv4Address,
}
impl MixtcpReqwestClient {
pub async fn new() -> Result<Self, MixtcpError> {
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<MixtcpResponse, MixtcpError> {
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<Vec<u8>, 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::Socket>(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<String, std::convert::Infallible> {
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<dyn std::error::Error>> {
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(())
}
mixtcp/examples/tls.rs
+270
View File
@@ -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<Self, MixtcpError> {
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<Vec<u8>, 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<dyn std::error::Error>> {
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::Socket>(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(())
}
mixtcp/src/bridge.rs
+120
View File
@@ -0,0 +1,120 @@
// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
// 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<Vec<u8>>,
/// Channel for sending incoming packets to device
rx_sender: mpsc::UnboundedSender<Vec<u8>>,
}
impl NymIprBridge {
pub fn new(
stream: IpMixStream,
tx_receiver: mpsc::UnboundedReceiver<Vec<u8>>,
rx_sender: mpsc::UnboundedSender<Vec<u8>>,
) -> 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(())
}
}
mixtcp/src/device.rs
+165
View File
@@ -0,0 +1,165 @@
// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
// 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<Vec<u8>>,
// Channel to send packets to the bridge task
tx_sender: mpsc::UnboundedSender<Vec<u8>>,
// Device capabilities
capabilities: DeviceCapabilities,
// Channel to receive packets from the bridge task
rx_receiver: mpsc::UnboundedReceiver<Vec<u8>>,
}
impl NymIprDevice {
pub fn new(
tx_sender: mpsc::UnboundedSender<Vec<u8>>,
rx_receiver: mpsc::UnboundedReceiver<Vec<u8>>,
) -> 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<Vec<u8>> {
self.tx_sender.clone()
}
/// Get the receiver for external use
pub fn rx_receiver(&self) -> mpsc::UnboundedReceiver<Vec<u8>> {
// 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<Self::TxToken<'_>> {
// 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<u8>,
}
impl RxToken for NymRxToken {
fn consume<R, F>(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<Vec<u8>>,
}
impl TxToken for NymTxToken {
fn consume<R, F>(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
}
}
mixtcp/src/error.rs
+52
View File
@@ -0,0 +1,52 @@
// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
// 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,
}
mixtcp/src/lib.rs
+49
View File
@@ -0,0 +1,49 @@
// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
// 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))
}