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

Rework error handling in tun device (#4146)

Browse Source 
* Rework error handling in tun device

* Extract out timeout constants

* Experiment with timeouts

* Update error msg

* try_send in one direction as hotfix for deadlock

* Downgrade some log from info to debug

* Update comment

* rustfmt
This commit is contained in:
Jon Häggblad
2023-11-17 09:52:05 +01:00
committed by GitHub
parent 808e3f0562
commit 1cdd8f6c08
5 changed files with 182 additions and 119 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/wireguard/src/active_peers.rs
+24 -4
View File
@@ -1,11 +1,14 @@
use std::net::SocketAddr;
use std::{net::SocketAddr, time::Duration};
use boringtun::x25519;
use dashmap::{
mapref::one::{Ref, RefMut},
DashMap,
};
use tokio::sync::mpsc::{self};
use tokio::{
sync::mpsc::{self},
time::{error::Elapsed, timeout},
};
use crate::event::Event;
@@ -14,9 +17,26 @@ use crate::event::Event;
pub struct PeerEventSender(mpsc::Sender<Event>);
pub(crate) struct PeerEventReceiver(mpsc::Receiver<Event>);
#[derive(thiserror::Error, Debug)]
pub enum PeerEventSenderError {
#[error("timeout")]
Timeout {
#[from]
source: Elapsed,
},
#[error("send failed: {source}")]
SendError {
#[from]
source: mpsc::error::SendError<Event>,
},
}
impl PeerEventSender {
pub(crate) async fn send(&self, event: Event) -> Result<(), mpsc::error::SendError<Event>> {
self.0.send(event).await
pub(crate) async fn send(&self, event: Event) -> Result<(), PeerEventSenderError> {
timeout(Duration::from_millis(1000), self.0.send(event))
.await?
.map_err(|err| err.into())
}
}
common/wireguard/src/platform/linux/tun_device.rs
+97 -79
View File
@@ -2,21 +2,64 @@ use std::{
collections::HashMap,
net::{IpAddr, Ipv4Addr},
sync::Arc,
time::Duration,
};
use etherparse::{InternetSlice, SlicedPacket};
use tap::TapFallible;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::{
io::{AsyncReadExt, AsyncWriteExt},
time::timeout,
};
use crate::{
active_peers::PeerEventSenderError,
event::Event,
tun_task_channel::{
tun_task_channel, tun_task_response_channel, TunTaskPayload, TunTaskResponseRx,
TunTaskResponseTx, TunTaskRx, TunTaskTx,
TunTaskResponseSendError, TunTaskResponseTx, TunTaskRx, TunTaskTx,
},
udp_listener::PeersByIp,
};
const MUTEX_LOCK_TIMEOUT_MS: u64 = 200;
const TUN_WRITE_TIMEOUT_MS: u64 = 1000;
#[derive(thiserror::Error, Debug)]
pub enum TunDeviceError {
#[error("timeout writing to tun device, dropping packet")]
TunWriteTimeout,
#[error("error writing to tun device: {source}")]
TunWriteError { source: std::io::Error },
#[error("failed forwarding packet to peer: {source}")]
ForwardToPeerFailed {
#[from]
source: PeerEventSenderError,
},
#[error("failed to forward responding packet with tag: {source}")]
ForwardNatResponseFailed {
#[from]
source: TunTaskResponseSendError,
},
#[error("unable to parse destination address from packet")]
UnableToParseDstAdddress,
#[error("unable to parse source address from packet")]
UnableToParseSrcAddress {
#[from]
source: etherparse::ReadError,
},
#[error("unable to parse source address from packet: ip header missing")]
UnableToParseSrcAddressIpHeaderMissing,
#[error("unable to lock peer mutex")]
FailedToLockPeer,
}
fn setup_tokio_tun_device(name: &str, address: Ipv4Addr, netmask: Ipv4Addr) -> tokio_tun::Tun {
log::info!("Creating TUN device with: address={address}, netmask={netmask}");
// Read MTU size from env variable NYM_MTU_SIZE, else default to 1420.
@@ -74,6 +117,17 @@ pub struct AllowedIpsInner {
peers_by_ip: Arc<tokio::sync::Mutex<PeersByIp>>,
}
impl AllowedIpsInner {
async fn lock(&self) -> Result<tokio::sync::MutexGuard<PeersByIp>, TunDeviceError> {
timeout(
Duration::from_millis(MUTEX_LOCK_TIMEOUT_MS),
self.peers_by_ip.as_ref().lock(),
)
.await
.map_err(|_| TunDeviceError::FailedToLockPeer)
}
}
pub struct NatInner {
nat_table: HashMap<IpAddr, u64>,
}
@@ -114,17 +168,13 @@ impl TunDevice {
}
// Send outbound packets out on the wild internet
async fn handle_tun_write(&mut self, data: TunTaskPayload) {
async fn handle_tun_write(&mut self, data: TunTaskPayload) -> Result<(), TunDeviceError> {
let (tag, packet) = data;
let Some(dst_addr) = boringtun::noise::Tunn::dst_address(&packet) else {
log::error!("Unable to parse dst_address in packet that was supposed to be written to tun device");
return;
};
let Some(src_addr) = parse_src_address(&packet) else {
log::error!("Unable to parse src_address in packet that was supposed to be written to tun device");
return;
};
log::info!(
let dst_addr = boringtun::noise::Tunn::dst_address(&packet)
.ok_or_else(|| TunDeviceError::UnableToParseDstAdddress)?;
let src_addr = parse_src_address(&packet)?;
log::debug!(
"iface: write Packet({src_addr} -> {dst_addr}, {} bytes)",
packet.len()
);
@@ -134,28 +184,21 @@ impl TunDevice {
nat_table.nat_table.insert(src_addr, tag);
}
tokio::time::timeout(
std::time::Duration::from_millis(1000),
timeout(
Duration::from_millis(TUN_WRITE_TIMEOUT_MS),
self.tun.write_all(&packet),
)
.await
.tap_err(|err| {
log::error!("iface: write error: {err}");
})
.ok();
.map_err(|_| TunDeviceError::TunWriteTimeout)?
.map_err(|err| TunDeviceError::TunWriteError { source: err })
}
// Receive reponse packets from the wild internet
async fn handle_tun_read(&self, packet: &[u8]) {
let Some(dst_addr) = boringtun::noise::Tunn::dst_address(packet) else {
log::error!("Unable to parse dst_address in packet that was read from tun device");
return;
};
let Some(src_addr) = parse_src_address(packet) else {
log::error!("Unable to parse src_address in packet that was read from tun device");
return;
};
log::info!(
async fn handle_tun_read(&self, packet: &[u8]) -> Result<(), TunDeviceError> {
let dst_addr = boringtun::noise::Tunn::dst_address(packet)
.ok_or(TunDeviceError::UnableToParseDstAdddress)?;
let src_addr = parse_src_address(packet)?;
log::debug!(
"iface: read Packet({src_addr} -> {dst_addr}, {} bytes)",
packet.len(),
);
@@ -165,46 +208,30 @@ impl TunDevice {
match self.routing_mode {
// This is how wireguard does it, by consulting the AllowedIPs table.
RoutingMode::AllowedIps(ref peers_by_ip) => {
let Ok(peers) = tokio::time::timeout(
std::time::Duration::from_millis(1000),
peers_by_ip.peers_by_ip.as_ref().lock(),
)
.await
else {
log::error!("Failed to lock peer");
return;
};
let peers = peers_by_ip.lock().await?;
if let Some(peer_tx) = peers.longest_match(dst_addr).map(|(_, tx)| tx) {
log::info!("Forward packet to wg tunnel");
tokio::time::timeout(
std::time::Duration::from_millis(1000),
peer_tx.send(Event::Ip(packet.to_vec().into())),
)
.await
.tap_err(|err| log::error!("Failed to forward packet to wg tunnel: {err}"))
.ok();
return;
log::debug!("Forward packet to wg tunnel");
return peer_tx
.send(Event::Ip(packet.to_vec().into()))
.await
.map_err(|err| err.into());
}
}
// But we can also do it by consulting the NAT table.
RoutingMode::Nat(ref nat_table) => {
if let Some(tag) = nat_table.nat_table.get(&dst_addr) {
log::info!("Forward packet with tag: {tag}");
tokio::time::timeout(
std::time::Duration::from_millis(1000),
self.tun_task_response_tx.send((*tag, packet.to_vec())),
)
.await
.tap_err(|err| log::error!("Failed to foward packet with tag: {err}"))
.ok();
return;
log::debug!("Forward packet with NAT tag: {tag}");
return self
.tun_task_response_tx
.try_send((*tag, packet.to_vec()))
.map_err(|err| err.into());
}
}
}
log::info!("No peer found, packet dropped");
Ok(())
}
pub async fn run(mut self) {
@@ -216,13 +243,9 @@ impl TunDevice {
len = self.tun.read(&mut buf) => match len {
Ok(len) => {
let packet = &buf[..len];
tokio::time::timeout(
std::time::Duration::from_millis(1000),
self.handle_tun_read(packet)
)
.await
.tap_err(|_err| log::error!("Failed: handle_tun_read timeout"))
.ok();
if let Err(err) = self.handle_tun_read(packet).await {
log::error!("iface: handle_tun_read failed: {err}")
}
},
Err(err) => {
log::info!("iface: read error: {err}");
@@ -231,13 +254,9 @@ impl TunDevice {
},
// Writing to the TUN device
Some(data) = self.tun_task_rx.recv() => {
tokio::time::timeout(
std::time::Duration::from_millis(1000),
self.handle_tun_write(data)
)
.await
.tap_err(|_err| log::error!("Failed: handle_tun_write timeout"))
.ok();
if let Err(err) = self.handle_tun_write(data).await {
log::error!("ifcae: handle_tun_write failed: {err}");
}
}
}
}
@@ -249,12 +268,11 @@ impl TunDevice {
}
}
fn parse_src_address(packet: &[u8]) -> Option<IpAddr> {
let headers = SlicedPacket::from_ip(packet)
.tap_err(|err| log::error!("Unable to parse IP packet: {err:?}"))
.ok()?;
Some(match headers.ip? {
InternetSlice::Ipv4(ip, _) => ip.source_addr().into(),
InternetSlice::Ipv6(ip, _) => ip.source_addr().into(),
})
fn parse_src_address(packet: &[u8]) -> Result<IpAddr, TunDeviceError> {
let headers = SlicedPacket::from_ip(packet)?;
match headers.ip {
Some(InternetSlice::Ipv4(ip, _)) => Ok(ip.source_addr().into()),
Some(InternetSlice::Ipv6(ip, _)) => Ok(ip.source_addr().into()),
None => Err(TunDeviceError::UnableToParseSrcAddressIpHeaderMissing),
}
}
common/wireguard/src/tun_task_channel.rs
+40 -12
View File
@@ -1,4 +1,9 @@
use tokio::sync::mpsc;
use std::time::Duration;
use tokio::sync::mpsc::{
self,
error::{SendError, SendTimeoutError, TrySendError},
};
pub(crate) type TunTaskPayload = (u64, Vec<u8>);
@@ -7,12 +12,13 @@ pub struct TunTaskTx(mpsc::Sender<TunTaskPayload>);
pub(crate) struct TunTaskRx(mpsc::Receiver<TunTaskPayload>);
impl TunTaskTx {
pub async fn send(
&self,
data: TunTaskPayload,
) -> Result<(), tokio::sync::mpsc::error::SendError<TunTaskPayload>> {
pub async fn send(&self, data: TunTaskPayload) -> Result<(), SendError<TunTaskPayload>> {
self.0.send(data).await
}
pub fn try_send(&self, data: TunTaskPayload) -> Result<(), TrySendError<TunTaskPayload>> {
self.0.try_send(data)
}
}
impl TunTaskRx {
@@ -22,20 +28,42 @@ impl TunTaskRx {
}
pub(crate) fn tun_task_channel() -> (TunTaskTx, TunTaskRx) {
let (tun_task_tx, tun_task_rx) = tokio::sync::mpsc::channel(16);
let (tun_task_tx, tun_task_rx) = tokio::sync::mpsc::channel(128);
(TunTaskTx(tun_task_tx), TunTaskRx(tun_task_rx))
}
const TUN_TASK_RESPONSE_SEND_TIMEOUT_MS: u64 = 1_000;
// Send responses back from the tun device back to the PacketRelayer
pub(crate) struct TunTaskResponseTx(mpsc::Sender<TunTaskPayload>);
pub struct TunTaskResponseRx(mpsc::Receiver<TunTaskPayload>);
#[derive(thiserror::Error, Debug)]
pub enum TunTaskResponseSendError {
#[error("failed to send tun response: {0}")]
SendTimeoutError(#[from] SendTimeoutError<TunTaskPayload>),
#[error("failed to send tun response: {0}")]
SendError(#[from] SendError<TunTaskPayload>),
#[error("failed to send tun response: {0}")]
TrySendError(#[from] TrySendError<TunTaskPayload>),
}
impl TunTaskResponseTx {
pub(crate) async fn send(
&self,
data: TunTaskPayload,
) -> Result<(), tokio::sync::mpsc::error::SendError<TunTaskPayload>> {
self.0.send(data).await
#[allow(unused)]
pub(crate) async fn send(&self, data: TunTaskPayload) -> Result<(), TunTaskResponseSendError> {
Ok(self
.0
.send_timeout(
data,
Duration::from_millis(TUN_TASK_RESPONSE_SEND_TIMEOUT_MS),
)
.await?)
}
pub(crate) fn try_send(&self, data: TunTaskPayload) -> Result<(), TunTaskResponseSendError> {
Ok(self.0.try_send(data)?)
}
}
@@ -46,7 +74,7 @@ impl TunTaskResponseRx {
}
pub(crate) fn tun_task_response_channel() -> (TunTaskResponseTx, TunTaskResponseRx) {
let (tun_task_tx, tun_task_rx) = tokio::sync::mpsc::channel(16);
let (tun_task_tx, tun_task_rx) = tokio::sync::mpsc::channel(128);
(
TunTaskResponseTx(tun_task_tx),
TunTaskResponseRx(tun_task_rx),
service-providers/ip-packet-router/src/error.rs
+5
View File
@@ -39,6 +39,11 @@ pub enum IpPacketRouterError {
#[error("parsed packet is missing transport header")]
PacketMissingTransportHeader,
#[error("failed to send packet to tun device: {source}")]
FailedToSendPacketToTun {
source: tokio::sync::mpsc::error::TrySendError<(u64, Vec<u8>)>,
},
#[error("the provided socket address, '{addr}' is not covered by the exit policy!")]
AddressNotCoveredByExitPolicy { addr: SocketAddr },
service-providers/ip-packet-router/src/lib.rs
+16 -24
View File
@@ -18,7 +18,6 @@ use nym_sdk::{
use nym_sphinx::receiver::ReconstructedMessage;
use nym_task::{connections::TransmissionLane, TaskClient, TaskHandle};
use request_filter::RequestFilter;
use tap::TapFallible;
use crate::config::BaseClientConfig;
@@ -202,7 +201,9 @@ impl IpPacketRouter {
},
msg = self.mixnet_client.next() => {
if let Some(msg) = msg {
self.on_message(msg).await.ok();
if let Err(err) = self.on_message(msg).await {
log::error!("Error handling mixnet message: {err}");
};
} else {
log::trace!("IpPacketRouter [main loop]: stopping since channel closed");
break;
@@ -225,13 +226,9 @@ impl IpPacketRouter {
let packet_type = None;
let input_message = InputMessage::new_regular(recipient, packet, lane, packet_type);
self.mixnet_client
.send(input_message)
.await
.tap_err(|err| {
log::error!("IpPacketRouter [main loop]: failed to send packet to mixnet: {err}");
})
.ok();
if let Err(err) = self.mixnet_client.send(input_message).await {
log::error!("IpPacketRouter [main loop]: failed to send packet to mixnet: {err}");
};
} else {
log::error!("NYM_CLIENT_ADDR not set or invalid");
}
@@ -251,7 +248,10 @@ impl IpPacketRouter {
&mut self,
reconstructed: ReconstructedMessage,
) -> Result<(), IpPacketRouterError> {
log::info!("Received message: {:?}", reconstructed.sender_tag);
log::debug!(
"Received message with sender_tag: {:?}",
reconstructed.sender_tag
);
// We don't forward packets that we are not able to parse. BUT, there might be a good
// reason to still forward them.
@@ -274,27 +274,19 @@ impl IpPacketRouter {
if let Some(dst) = dst {
if !self.request_filter.check_address(&dst).await {
log::warn!("Failed filter check: {dst}");
// TODO: send back a response here
// TODO: we could consider sending back a response here
return Err(IpPacketRouterError::AddressFailedFilterCheck { addr: dst });
}
} else {
// TODO: are we always allowing packets without port numbers?
log::warn!(
"Ignoring filter check for packet without port number! (TODO: is this correct?)"
);
// TODO: we should also filter packets without port number
log::warn!("Ignoring filter check for packet without port number! TODO!");
}
// TODO: set the tag correctly. Can we just reuse sender_tag?
let peer_tag = 0;
self.tun_task_tx
.send((peer_tag, reconstructed.message))
.await
.tap_err(|err| {
log::error!("Failed to send packet to tun device: {err}");
})
.ok();
.try_send((peer_tag, reconstructed.message))
.map_err(|err| IpPacketRouterError::FailedToSendPacketToTun { source: err })?;
Ok(())
}
@@ -316,7 +308,7 @@ 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::Icmpv4(_)) => ("icpv4", None),
Some(etherparse::TransportSlice::Icmpv4(_)) => ("icmpv4", None),
Some(etherparse::TransportSlice::Icmpv6(_)) => ("icmpv6", None),
Some(etherparse::TransportSlice::Unknown(_)) => ("unknown", None),
None => {