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

Sort out task flow in ip packet router (#4200)

Browse Source 
* Create separate tun listener task

* log tweaks

* typo

* typo in log

* Add sync event

* typo in log

* Add note

* Switch to unbounded channel

* tweak log

* Handle tun device directly in ip-packet-router

* clippy

* update comment

* Fix mac build

* allow unused imports on non-linux

* listen for shutdown in tun listener

* log tweaks
This commit is contained in:
Jon Häggblad
2023-11-30 11:40:43 +01:00
committed by GitHub
parent 03b4c87115
commit b1fa70a232
6 changed files with 136 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+3 -2
View File
@@ -6652,6 +6652,7 @@ dependencies = [
"tap",
"thiserror",
"tokio",
"tokio-tun",
"url",
]
@@ -10944,9 +10945,9 @@ dependencies = [
[[package]]
name = "tokio-tun"
version = "0.9.1"
version = "0.11.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c4a67d1405a577ba1f4cd61f46608f1db2cadbb6a9549c3fc2eed7f1195393c9"
checksum = "bf2efaf33e86779a3a68b1f1d6e9e13a346c1c75ee3cab7a4293235c463b2668"
dependencies = [
"libc",
"nix 0.27.1",
common/tun/Cargo.toml
+1 -1
View File
@@ -18,4 +18,4 @@ log.workspace = true
nym-wireguard-types = { path = "../wireguard-types", optional = true }
[target.'cfg(target_os = "linux")'.dependencies]
tokio-tun = "0.9.0"
tokio-tun = "0.11.2"
common/tun/src/linux/tun_device.rs
+16 -10
View File
@@ -112,15 +112,7 @@ impl TunDevice {
routing_mode: RoutingMode,
config: TunDeviceConfig,
) -> (Self, TunTaskTx, TunTaskResponseRx) {
let TunDeviceConfig {
base_name,
ip,
netmask,
} = config;
let name = format!("{base_name}%d");
let tun = setup_tokio_tun_device(&name, ip, netmask);
log::info!("Created TUN device: {}", tun.name());
let tun = Self::new_device_only(config);
// Channels to communicate with the other tasks
let (tun_task_tx, tun_task_rx) = tun_task_channel();
@@ -136,6 +128,19 @@ impl TunDevice {
(tun_device, tun_task_tx, tun_task_response_rx)
}
pub fn new_device_only(config: TunDeviceConfig) -> tokio_tun::Tun {
let TunDeviceConfig {
base_name,
ip,
netmask,
} = config;
let name = format!("{base_name}%d");
let tun = setup_tokio_tun_device(&name, ip, netmask);
log::info!("Created TUN device: {}", tun.name());
tun
}
// Send outbound packets out on the wild internet
async fn handle_tun_write(&mut self, data: TunTaskPayload) -> Result<(), TunDeviceError> {
let (tag, packet) = data;
@@ -182,6 +187,7 @@ impl TunDevice {
}
RoutingMode::Passthrough => {
// TODO: skip the parsing at the top of the function
log::debug!("Forward packet without checking anything");
return self
.tun_task_response_tx
@@ -215,7 +221,7 @@ impl TunDevice {
// Writing to the TUN device
Some(data) = self.tun_task_rx.recv() => {
if let Err(err) = self.handle_tun_write(data).await {
log::error!("ifcae: handle_tun_write failed: {err}");
log::error!("iface: handle_tun_write failed: {err}");
}
}
}
service-providers/ip-packet-router/Cargo.toml
+3
View File
@@ -35,3 +35,6 @@ tap.workspace = true
thiserror = { workspace = true }
tokio = { workspace = true, features = ["rt-multi-thread", "net", "io-util"] }
url.workspace = true
[target.'cfg(target_os = "linux")'.dependencies]
tokio-tun = "0.11.2"
service-providers/ip-packet-router/src/error.rs
+3
View File
@@ -50,6 +50,9 @@ pub enum IpPacketRouterError {
source: tokio::sync::mpsc::error::TrySendError<(u64, Vec<u8>)>,
},
#[error("failed to write packet to tun")]
FailedToWritePacketToTun,
#[error("the provided socket address, '{addr}' is not covered by the exit policy!")]
AddressNotCoveredByExitPolicy { addr: SocketAddr },
service-providers/ip-packet-router/src/lib.rs
+110 -28
View File
@@ -1,4 +1,5 @@
#![cfg_attr(not(target_os = "linux"), allow(dead_code))]
#![cfg_attr(not(target_os = "linux"), allow(unused_imports))]
use std::{
collections::HashMap,
@@ -24,6 +25,8 @@ use nym_sdk::{
use nym_sphinx::receiver::ReconstructedMessage;
use nym_task::{connections::TransmissionLane, TaskClient, TaskHandle};
use request_filter::RequestFilter;
#[cfg(target_os = "linux")]
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use crate::config::BaseClientConfig;
@@ -154,11 +157,21 @@ impl IpPacketRouterBuilder {
ip: TUN_DEVICE_ADDRESS.parse().unwrap(),
netmask: TUN_DEVICE_NETMASK.parse().unwrap(),
};
let (tun, tun_task_tx, tun_task_response_rx) = nym_tun::tun_device::TunDevice::new(
nym_tun::tun_device::RoutingMode::new_passthrough(),
config,
);
tun.start();
let (tun_reader, tun_writer) =
tokio::io::split(nym_tun::tun_device::TunDevice::new_device_only(config));
// Channel used by the IpPacketRouter to signal connected and disconnected clients to the
// TunListener
let (connected_client_tx, connected_client_rx) = tokio::sync::mpsc::unbounded_channel();
let tun_listener = TunListener {
tun_reader,
mixnet_client_sender: mixnet_client.split_sender(),
task_client: task_handle.get_handle(),
connected_clients: Default::default(),
connected_client_rx,
};
tun_listener.start();
let request_filter = request_filter::RequestFilter::new(&self.config).await?;
request_filter.start_update_tasks().await;
@@ -166,11 +179,11 @@ impl IpPacketRouterBuilder {
let ip_packet_router_service = IpPacketRouter {
_config: self.config,
request_filter: request_filter.clone(),
tun_task_tx,
tun_task_response_rx,
tun_writer,
mixnet_client,
task_handle,
connected_clients: Default::default(),
connected_client_tx,
};
log::info!("The address of this client is: {self_address}");
@@ -190,16 +203,16 @@ impl IpPacketRouterBuilder {
}
}
#[cfg_attr(not(target_os = "linux"), allow(unused))]
#[cfg(target_os = "linux")]
struct IpPacketRouter {
_config: Config,
request_filter: request_filter::RequestFilter,
tun_task_tx: nym_tun::tun_task_channel::TunTaskTx,
tun_task_response_rx: nym_tun::tun_task_channel::TunTaskResponseRx,
tun_writer: tokio::io::WriteHalf<tokio_tun::Tun>,
mixnet_client: nym_sdk::mixnet::MixnetClient,
task_handle: TaskHandle,
connected_clients: HashMap<IpAddr, ConnectedClient>,
connected_client_tx: tokio::sync::mpsc::UnboundedSender<ConnectedClientEvent>,
}
struct ConnectedClient {
@@ -207,7 +220,7 @@ struct ConnectedClient {
last_activity: std::time::Instant,
}
#[cfg_attr(not(target_os = "linux"), allow(unused))]
#[cfg(target_os = "linux")]
impl IpPacketRouter {
async fn on_static_connect_request(
&mut self,
@@ -254,6 +267,12 @@ impl IpPacketRouter {
last_activity: std::time::Instant::now(),
},
);
self.connected_client_tx
.send(ConnectedClientEvent::Connect(
requested_ip,
Box::new(reply_to),
))
.unwrap();
Ok(Some(IpPacketResponse::new_static_connect_success(
request_id, reply_to,
)))
@@ -332,6 +351,9 @@ impl IpPacketRouter {
last_activity: std::time::Instant::now(),
},
);
self.connected_client_tx
.send(ConnectedClientEvent::Connect(new_ip, Box::new(reply_to)))
.unwrap();
Ok(Some(IpPacketResponse::new_dynamic_connect_success(
request_id, reply_to, new_ip,
)))
@@ -341,7 +363,7 @@ impl IpPacketRouter {
&mut self,
data_request: nym_ip_packet_requests::DataRequest,
) -> Result<Option<IpPacketResponse>, IpPacketRouterError> {
log::info!("Received data request");
log::trace!("Received data request");
// We don't forward packets that we are not able to parse. BUT, there might be a good
// reason to still forward them.
@@ -382,9 +404,11 @@ impl IpPacketRouter {
}
// TODO: consider changing from Vec<u8> to bytes::Bytes?
self.tun_task_tx
.try_send((0, data_request.ip_packet.into()))
.map_err(|err| IpPacketRouterError::FailedToSendPacketToTun { source: err })?;
let packet = data_request.ip_packet;
self.tun_writer
.write_all(&packet)
.await
.map_err(|_| IpPacketRouterError::FailedToWritePacketToTun)?;
Ok(None)
}
@@ -444,6 +468,7 @@ impl IpPacketRouter {
for ip in inactive_clients {
log::info!("Disconnect inactive client: {ip}");
self.connected_clients.remove(&ip);
self.connected_client_tx.send(ConnectedClientEvent::Disconnect(ip)).unwrap();
}
},
msg = self.mixnet_client.next() => {
@@ -479,9 +504,52 @@ impl IpPacketRouter {
break;
};
},
packet = self.tun_task_response_rx.recv() => {
if let Some((_tag, packet)) = packet {
let Some(dst_addr) = parse_dst_addr(&packet) else {
}
}
log::debug!("IpPacketRouter: stopping");
Ok(())
}
}
// Reads packet from TUN and writes to mixnet client
#[cfg(target_os = "linux")]
struct TunListener {
tun_reader: tokio::io::ReadHalf<tokio_tun::Tun>,
mixnet_client_sender: nym_sdk::mixnet::MixnetClientSender,
task_client: TaskClient,
// A mirror of the one in IpPacketRouter
connected_clients: HashMap<IpAddr, ConnectedClient>,
connected_client_rx: tokio::sync::mpsc::UnboundedReceiver<ConnectedClientEvent>,
}
#[cfg(target_os = "linux")]
impl TunListener {
async fn run(mut self) -> Result<(), IpPacketRouterError> {
let mut buf = [0u8; 65535];
while !self.task_client.is_shutdown() {
tokio::select! {
_ = self.task_client.recv() => {
log::trace!("TunListener: received shutdown");
},
event = self.connected_client_rx.recv() => match event {
Some(ConnectedClientEvent::Connect(ip, nym_addr)) => {
log::trace!("Connect client: {ip}");
self.connected_clients.insert(ip, ConnectedClient {
nym_address: *nym_addr,
last_activity: std::time::Instant::now(),
});
},
Some(ConnectedClientEvent::Disconnect(ip)) => {
log::trace!("Disconnect client: {ip}");
self.connected_clients.remove(&ip);
},
None => {},
},
len = self.tun_reader.read(&mut buf) => match len {
Ok(len) => {
let Some(dst_addr) = parse_dst_addr(&buf[..len]) else {
log::warn!("Failed to parse packet");
continue;
};
@@ -491,6 +559,7 @@ impl IpPacketRouter {
if let Some(recipient) = recipient {
let lane = TransmissionLane::General;
let packet_type = None;
let packet = buf[..len].to_vec();
let response_packet = IpPacketResponse::new_ip_packet(packet.into()).to_bytes();
let Ok(response_packet) = response_packet else {
log::error!("Failed to serialize response packet");
@@ -498,23 +567,36 @@ impl IpPacketRouter {
};
let input_message = InputMessage::new_regular(recipient, response_packet, lane, packet_type);
if let Err(err) = self.mixnet_client.send(input_message).await {
log::error!("IpPacketRouter [main loop]: failed to send packet to mixnet: {err}");
if let Err(err) = self.mixnet_client_sender.send(input_message).await {
log::error!("TunListener: failed to send packet to mixnet: {err}");
};
} else {
log::error!("IpPacketRouter [main loop]: no nym-address recipient for packet");
log::info!("No registered nym-address for packet - dropping");
}
} else {
log::trace!("IpPacketRouter [main loop]: stopping since channel closed");
break;
},
Err(err) => {
log::warn!("iface: read error: {err}");
// break;
}
}
}
}
log::info!("IpPacketRouter: stopping");
log::debug!("TunListener: stopping");
Ok(())
}
fn start(self) {
tokio::spawn(async move {
if let Err(err) = self.run().await {
log::error!("tun listener router has failed: {err}")
}
});
}
}
enum ConnectedClientEvent {
Disconnect(IpAddr),
Connect(IpAddr, Box<Recipient>),
}
struct ParsedPacket<'a> {
@@ -531,8 +613,8 @@ fn parse_packet(packet: &[u8]) -> Result<ParsedPacket, IpPacketRouterError> {
})?;
let (packet_type, dst_port) = match headers.transport {
Some(etherparse::TransportSlice::Udp(header)) => ("ipv4", Some(header.destination_port())),
Some(etherparse::TransportSlice::Tcp(header)) => ("ipv6", Some(header.destination_port())),
Some(etherparse::TransportSlice::Udp(header)) => ("udp", Some(header.destination_port())),
Some(etherparse::TransportSlice::Tcp(header)) => ("tcp", Some(header.destination_port())),
Some(etherparse::TransportSlice::Icmpv4(_)) => ("icmpv4", None),
Some(etherparse::TransportSlice::Icmpv6(_)) => ("icmpv6", None),
Some(etherparse::TransportSlice::Unknown(_)) => ("unknown", None),