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

Merge pull request #4068 from nymtech/jon/ip-forwarder-tun-device

Browse Source 
Create IpPacketRouter
This commit is contained in:
Jon Häggblad
2023-11-03 13:34:37 +01:00
committed by GitHub
parent ec8c67e67d 807e7e588f
commit 9f2bff2d16
13 changed files with 267 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+5
View File
@@ -6649,6 +6649,7 @@ dependencies = [
name = "nym-ip-packet-router"
version = "0.1.0"
dependencies = [
"etherparse",
"futures",
"log",
"nym-bin-common",
@@ -6658,9 +6659,13 @@ dependencies = [
"nym-service-providers-common",
"nym-sphinx",
"nym-task",
"nym-wireguard",
"nym-wireguard-types",
"serde",
"serde_json",
"tap",
"thiserror",
"tokio",
]
[[package]]
common/wireguard/src/lib.rs
+14 -6
View File
@@ -11,7 +11,7 @@ mod packet_relayer;
mod platform;
mod registered_peers;
mod setup;
mod tun_task_channel;
pub mod tun_task_channel;
mod udp_listener;
mod wg_tunnel;
@@ -20,7 +20,7 @@ use std::sync::Arc;
// Currently the module related to setting up the virtual network device is platform specific.
#[cfg(target_os = "linux")]
use platform::linux::tun_device;
pub use platform::linux::tun_device;
/// Start wireguard UDP listener and TUN device
///
@@ -32,16 +32,24 @@ pub async fn start_wireguard(
task_client: nym_task::TaskClient,
gateway_client_registry: Arc<GatewayClientRegistry>,
) -> Result<(), Box<dyn std::error::Error + Send + Sync + 'static>> {
// We can either index peers by their IP like standard wireguard
// TODO: make this configurable
// We can optionally index peers by their IP like standard wireguard. If we don't then we do
// plain NAT where we match incoming destination IP with outgoing source IP.
let peers_by_ip = Arc::new(tokio::sync::Mutex::new(network_table::NetworkTable::new()));
// ... or by their tunnel tag, which is a random number assigned to them
let peers_by_tag = Arc::new(tokio::sync::Mutex::new(wg_tunnel::PeersByTag::new()));
// Alternative 1:
let routing_mode = tun_device::RoutingMode::new_allowed_ips(peers_by_ip.clone());
// Alternative 2:
//let routing_mode = tun_device::RoutingMode::new_nat();
// Start the tun device that is used to relay traffic outbound
let (tun, tun_task_tx, tun_task_response_rx) = tun_device::TunDevice::new(peers_by_ip.clone());
let (tun, tun_task_tx, tun_task_response_rx) = tun_device::TunDevice::new(routing_mode);
tun.start();
// We also index peers by a tag
let peers_by_tag = Arc::new(tokio::sync::Mutex::new(wg_tunnel::PeersByTag::new()));
// If we want to have the tun device on a separate host, it's the tun_task and
// tun_task_response channels that needs to be sent over the network to the host where the tun
// device is running.
common/wireguard/src/platform/linux/mod.rs
+1 -1
View File
@@ -1 +1 @@
pub(crate) mod tun_device;
pub mod tun_device;
common/wireguard/src/platform/linux/tun_device.rs
+55 -27
View File
@@ -42,17 +42,42 @@ pub struct TunDevice {
// And when we get replies, this is where we should send it
tun_task_response_tx: TunTaskResponseTx,
routing_mode: RoutingMode,
}
pub enum RoutingMode {
// The routing table, as how wireguard does it
peers_by_ip: Arc<tokio::sync::Mutex<PeersByIp>>,
AllowedIps(AllowedIpsInner),
// This is an alternative to the routing table, where we just match outgoing source IP with
// incoming destination IP.
Nat(NatInner),
}
impl RoutingMode {
pub fn new_nat() -> Self {
RoutingMode::Nat(NatInner {
nat_table: HashMap::new(),
})
}
pub fn new_allowed_ips(peers_by_ip: Arc<tokio::sync::Mutex<PeersByIp>>) -> Self {
RoutingMode::AllowedIps(AllowedIpsInner { peers_by_ip })
}
}
pub struct AllowedIpsInner {
peers_by_ip: Arc<tokio::sync::Mutex<PeersByIp>>,
}
pub struct NatInner {
nat_table: HashMap<IpAddr, u64>,
}
impl TunDevice {
pub fn new(
peers_by_ip: Arc<tokio::sync::Mutex<PeersByIp>>,
routing_mode: RoutingMode,
// peers_by_ip: Option<Arc<tokio::sync::Mutex<PeersByIp>>>,
) -> (Self, TunTaskTx, TunTaskResponseRx) {
let tun = setup_tokio_tun_device(
format!("{TUN_BASE_NAME}%d").as_str(),
@@ -69,8 +94,7 @@ impl TunDevice {
tun_task_rx,
tun_task_response_tx,
tun,
peers_by_ip,
nat_table: HashMap::new(),
routing_mode,
};
(tun_device, tun_task_tx, tun_task_response_rx)
@@ -93,7 +117,9 @@ impl TunDevice {
);
// TODO: expire old entries
self.nat_table.insert(src_addr, tag);
if let RoutingMode::Nat(nat_table) = &mut self.routing_mode {
nat_table.nat_table.insert(src_addr, tag);
}
self.tun
.write_all(&packet)
@@ -121,30 +147,32 @@ impl TunDevice {
// Route packet to the correct peer.
// This is how wireguard does it, by consulting the AllowedIPs table.
if false {
let peers = self.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");
peer_tx
.send(Event::Ip(packet.to_vec().into()))
.await
.tap_err(|err| log::error!("{err}"))
.ok();
return;
match self.routing_mode {
// This is how wireguard does it, by consulting the AllowedIPs table.
RoutingMode::AllowedIps(ref peers_by_ip) => {
let peers = peers_by_ip.peers_by_ip.as_ref().lock().await;
if let Some(peer_tx) = peers.longest_match(dst_addr).map(|(_, tx)| tx) {
log::info!("Forward packet to wg tunnel");
peer_tx
.send(Event::Ip(packet.to_vec().into()))
.await
.tap_err(|err| log::error!("{err}"))
.ok();
return;
}
}
}
// But we do it by consulting the NAT table.
{
if let Some(tag) = self.nat_table.get(&dst_addr) {
log::info!("Forward packet to wg tunnel with tag: {tag}");
self.tun_task_response_tx
.send((*tag, packet.to_vec()))
.await
.tap_err(|err| log::error!("{err}"))
.ok();
return;
// But we 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 to wg tunnel with tag: {tag}");
self.tun_task_response_tx
.send((*tag, packet.to_vec()))
.await
.tap_err(|err| log::error!("{err}"))
.ok();
return;
}
}
}
common/wireguard/src/tun_task_channel.rs
+2 -2
View File
@@ -7,7 +7,7 @@ pub struct TunTaskTx(mpsc::Sender<TunTaskPayload>);
pub(crate) struct TunTaskRx(mpsc::Receiver<TunTaskPayload>);
impl TunTaskTx {
pub(crate) async fn send(
pub async fn send(
&self,
data: TunTaskPayload,
) -> Result<(), tokio::sync::mpsc::error::SendError<TunTaskPayload>> {
@@ -40,7 +40,7 @@ impl TunTaskResponseTx {
}
impl TunTaskResponseRx {
pub(crate) async fn recv(&mut self) -> Option<TunTaskPayload> {
pub async fn recv(&mut self) -> Option<TunTaskPayload> {
self.0.recv().await
}
}
gateway/src/error.rs
+2 -2
View File
@@ -2,7 +2,7 @@
// SPDX-License-Identifier: Apache-2.0
use crate::node::storage::error::StorageError;
use nym_ip_packet_router::error::IpForwarderError;
use nym_ip_packet_router::error::IpPacketRouterError;
use nym_network_requester::error::{ClientCoreError, NetworkRequesterError};
use nym_validator_client::nyxd::error::NyxdError;
use nym_validator_client::nyxd::AccountId;
@@ -110,7 +110,7 @@ pub(crate) enum GatewayError {
#[error("there was an issue with the local ip packet router: {source}")]
IpPacketRouterFailure {
#[from]
source: IpForwarderError,
source: IpPacketRouterError,
},
#[error("failed to startup local network requester")]
gateway/src/node/mod.rs
+6 -5
View File
@@ -346,11 +346,12 @@ impl<St> Gateway<St> {
// TODO: well, wire it up internally to gateway traffic, shutdowns, etc.
let (on_start_tx, on_start_rx) = oneshot::channel();
let mut ip_builder = nym_ip_packet_router::IpForwarderBuilder::new(ip_opts.config.clone())
.with_shutdown(shutdown)
.with_custom_gateway_transceiver(Box::new(transceiver))
.with_wait_for_gateway(true)
.with_on_start(on_start_tx);
let mut ip_builder =
nym_ip_packet_router::IpPacketRouterBuilder::new(ip_opts.config.clone())
.with_shutdown(shutdown)
.with_custom_gateway_transceiver(Box::new(transceiver))
.with_wait_for_gateway(true)
.with_on_start(on_start_tx);
if let Some(custom_mixnet) = &ip_opts.custom_mixnet_path {
ip_builder = ip_builder.with_stored_topology(custom_mixnet)?
service-providers/ip-packet-router/Cargo.toml
+5
View File
@@ -9,6 +9,7 @@ edition.workspace = true
license.workspace = true
[dependencies]
etherparse = "0.13.0"
futures = { workspace = true }
log = { workspace = true }
nym-bin-common = { path = "../../common/bin-common" }
@@ -18,6 +19,10 @@ nym-sdk = { path = "../../sdk/rust/nym-sdk" }
nym-service-providers-common = { path = "../common" }
nym-sphinx = { path = "../../common/nymsphinx" }
nym-task = { path = "../../common/task" }
nym-wireguard = { path = "../../common/wireguard" }
nym-wireguard-types = { path = "../../common/wireguard-types" }
serde = { workspace = true, features = ["derive"] }
serde_json = { workspace = true }
tap.workspace = true
thiserror = { workspace = true }
tokio = { workspace = true, features = ["rt-multi-thread", "net", "io-util"] }
service-providers/ip-packet-router/src/config/mod.rs
+11 -11
View File
@@ -12,28 +12,28 @@ use std::{
path::{Path, PathBuf},
};
use crate::config::persistence::IpForwarderPaths;
use crate::config::persistence::IpPacketRouterPaths;
use self::template::CONFIG_TEMPLATE;
mod persistence;
mod template;
const DEFAULT_IP_FORWARDERS_DIR: &str = "ip-forwarder";
const DEFAULT_IP_PACKET_ROUTER_DIR: &str = "ip-packet-router";
/// Derive default path to ip forwarder's config directory.
/// It should get resolved to `$HOME/.nym/service-providers/ip-forwareder/<id>/config`
/// Derive default path to ip packet routers' config directory.
/// It should get resolved to `$HOME/.nym/service-providers/ip-packet-router/<id>/config`
pub fn default_config_directory<P: AsRef<Path>>(id: P) -> PathBuf {
must_get_home()
.join(NYM_DIR)
.join(DEFAULT_SERVICE_PROVIDERS_DIR)
.join(DEFAULT_IP_FORWARDERS_DIR)
.join(DEFAULT_IP_PACKET_ROUTER_DIR)
.join(id)
.join(DEFAULT_CONFIG_DIR)
}
/// Derive default path to ip forwarder's config file.
/// It should get resolved to `$HOME/.nym/service-providers/ip-forwarder/<id>/config/config.toml`
/// Derive default path to ip packet routers' config file.
/// It should get resolved to `$HOME/.nym/service-providers/ip-packet-router/<id>/config/config.toml`
pub fn default_config_filepath<P: AsRef<Path>>(id: P) -> PathBuf {
default_config_directory(id).join(DEFAULT_CONFIG_FILENAME)
}
@@ -44,7 +44,7 @@ pub fn default_data_directory<P: AsRef<Path>>(id: P) -> PathBuf {
must_get_home()
.join(NYM_DIR)
.join(DEFAULT_SERVICE_PROVIDERS_DIR)
.join(DEFAULT_IP_FORWARDERS_DIR)
.join(DEFAULT_IP_PACKET_ROUTER_DIR)
.join(id)
.join(DEFAULT_DATA_DIR)
}
@@ -55,7 +55,7 @@ pub struct Config {
#[serde(flatten)]
pub base: BaseClientConfig,
pub storage_paths: IpForwarderPaths,
pub storage_paths: IpPacketRouterPaths,
pub logging: LoggingSettings,
}
@@ -70,13 +70,13 @@ impl Config {
pub fn new<S: AsRef<str>>(id: S) -> Self {
Config {
base: BaseClientConfig::new(id.as_ref(), env!("CARGO_PKG_VERSION")),
storage_paths: IpForwarderPaths::new_base(default_data_directory(id.as_ref())),
storage_paths: IpPacketRouterPaths::new_base(default_data_directory(id.as_ref())),
logging: Default::default(),
}
}
pub fn with_data_directory<P: AsRef<Path>>(mut self, data_directory: P) -> Self {
self.storage_paths = IpForwarderPaths::new_base(data_directory);
self.storage_paths = IpPacketRouterPaths::new_base(data_directory);
self
}
service-providers/ip-packet-router/src/config/persistence.rs
+4 -4
View File
@@ -8,21 +8,21 @@ use std::path::{Path, PathBuf};
pub const DEFAULT_DESCRIPTION_FILENAME: &str = "description.toml";
#[derive(Debug, Deserialize, PartialEq, Eq, Serialize, Clone)]
pub struct IpForwarderPaths {
pub struct IpPacketRouterPaths {
#[serde(flatten)]
pub common_paths: CommonClientPaths,
/// Location of the file containing our description
pub ip_forwarder_description: PathBuf,
pub ip_packet_router_description: PathBuf,
}
impl IpForwarderPaths {
impl IpPacketRouterPaths {
pub fn new_base<P: AsRef<Path>>(base_data_directory: P) -> Self {
let base_dir = base_data_directory.as_ref();
Self {
common_paths: CommonClientPaths::new_base(base_dir),
ip_forwarder_description: base_dir.join(DEFAULT_DESCRIPTION_FILENAME),
ip_packet_router_description: base_dir.join(DEFAULT_DESCRIPTION_FILENAME),
}
}
}
service-providers/ip-packet-router/src/config/template.rs
+1 -1
View File
@@ -76,7 +76,7 @@ allowed_list_location = '{{ storage_paths.allowed_list_location }}'
unknown_list_location = '{{ storage_paths.unknown_list_location }}'
# Path to file containing description of this network-requester.
ip_forwarder_description = '{{ storage_paths.ip_forwarder_description }}'
ip_packet_router_description = '{{ storage_paths.ip_packet_router_description }}'
##### logging configuration options #####
service-providers/ip-packet-router/src/error.rs
+8 -2
View File
@@ -1,7 +1,7 @@
pub use nym_client_core::error::ClientCoreError;
#[derive(thiserror::Error, Debug)]
pub enum IpForwarderError {
pub enum IpPacketRouterError {
#[error("I/O error: {0}")]
IoError(#[from] std::io::Error),
@@ -12,7 +12,7 @@ pub enum IpForwarderError {
FailedToLoadConfig(String),
// TODO: add more details here
#[error("Failed to validate the loaded config")]
#[error("failed to validate the loaded config")]
ConfigValidationFailure,
#[error("failed local version check, client and config mismatch")]
@@ -26,4 +26,10 @@ pub enum IpForwarderError {
#[error("the entity wrapping the network requester has disconnected")]
DisconnectedParent,
#[error("failed to parse incoming packet: {source}")]
PacketParseFailed { source: etherparse::ReadError },
#[error("parsed packet is missing IP header")]
PacketMissingHeader,
}
service-providers/ip-packet-router/src/lib.rs
+153 -17
View File
@@ -1,13 +1,18 @@
use std::path::Path;
use std::{net::IpAddr, path::Path};
use error::IpForwarderError;
use futures::channel::oneshot;
use error::IpPacketRouterError;
use futures::{channel::oneshot, StreamExt};
use nym_client_core::{
client::mix_traffic::transceiver::GatewayTransceiver,
config::disk_persistence::CommonClientPaths, HardcodedTopologyProvider, TopologyProvider,
};
use nym_sdk::{mixnet::Recipient, NymNetworkDetails};
use nym_task::{TaskClient, TaskHandle};
use nym_sdk::{
mixnet::{InputMessage, MixnetMessageSender, Recipient},
NymNetworkDetails,
};
use nym_sphinx::receiver::ReconstructedMessage;
use nym_task::{connections::TransmissionLane, TaskClient, TaskHandle};
use tap::TapFallible;
use crate::config::BaseClientConfig;
@@ -27,7 +32,8 @@ impl OnStartData {
}
}
pub struct IpForwarderBuilder {
pub struct IpPacketRouterBuilder {
#[allow(unused)]
config: Config,
wait_for_gateway: bool,
custom_topology_provider: Option<Box<dyn TopologyProvider + Send + Sync>>,
@@ -36,7 +42,7 @@ pub struct IpForwarderBuilder {
on_start: Option<oneshot::Sender<OnStartData>>,
}
impl IpForwarderBuilder {
impl IpPacketRouterBuilder {
pub fn new(config: Config) -> Self {
Self {
config,
@@ -87,20 +93,26 @@ impl IpForwarderBuilder {
pub fn with_stored_topology<P: AsRef<Path>>(
mut self,
file: P,
) -> Result<Self, IpForwarderError> {
) -> Result<Self, IpPacketRouterError> {
self.custom_topology_provider =
Some(Box::new(HardcodedTopologyProvider::new_from_file(file)?));
Ok(self)
}
pub async fn run_service_provider(self) -> Result<(), IpForwarderError> {
#[cfg(not(target_os = "linux"))]
pub async fn run_service_provider(self) -> Result<(), IpPacketRouterError> {
todo!("service provider is not yet supported on this platform")
}
#[cfg(target_os = "linux")]
pub async fn run_service_provider(self) -> Result<(), IpPacketRouterError> {
// Used to notify tasks to shutdown. Not all tasks fully supports this (yet).
let shutdown: TaskHandle = self.shutdown.map(Into::into).unwrap_or_default();
let task_handle: TaskHandle = self.shutdown.map(Into::into).unwrap_or_default();
// Connect to the mixnet
let mixnet_client = create_mixnet_client(
&self.config.base,
shutdown.get_handle().named("nym_sdk::MixnetClient"),
task_handle.get_handle().named("nym_sdk::MixnetClient"),
self.custom_gateway_transceiver,
self.custom_topology_provider,
self.wait_for_gateway,
@@ -110,17 +122,140 @@ impl IpForwarderBuilder {
let self_address = *mixnet_client.nym_address();
// Create the TUN device that we interact with the rest of the world with
let (tun, tun_task_tx, tun_task_response_rx) = nym_wireguard::tun_device::TunDevice::new(
nym_wireguard::tun_device::RoutingMode::new_nat(),
);
tun.start();
let ip_packet_router_service = IpPacketRouter {
_config: self.config,
// tun,
tun_task_tx,
tun_task_response_rx,
mixnet_client,
task_handle,
};
log::info!("The address of this client is: {self_address}");
log::info!("All systems go. Press CTRL-C to stop the server.");
if let Some(on_start) = self.on_start {
if on_start.send(OnStartData::new(self_address)).is_err() {
// the parent has dropped the channel before receiving the response
return Err(IpForwarderError::DisconnectedParent);
return Err(IpPacketRouterError::DisconnectedParent);
}
}
todo!();
ip_packet_router_service.run().await
}
}
#[allow(unused)]
struct IpPacketRouter {
_config: Config,
// tun: nym_wireguard::tun_device::TunDevice,
tun_task_tx: nym_wireguard::tun_task_channel::TunTaskTx,
tun_task_response_rx: nym_wireguard::tun_task_channel::TunTaskResponseRx,
mixnet_client: nym_sdk::mixnet::MixnetClient,
task_handle: TaskHandle,
}
#[allow(unused)]
impl IpPacketRouter {
async fn run(mut self) -> Result<(), IpPacketRouterError> {
let mut task_client = self.task_handle.fork("main_loop");
while !task_client.is_shutdown() {
tokio::select! {
_ = task_client.recv() => {
log::debug!("IpPacketRouter [main loop]: received shutdown");
},
msg = self.mixnet_client.next() => {
if let Some(msg) = msg {
self.on_message(msg).await.ok();
} else {
log::trace!("IpPacketRouter [main loop]: stopping since channel closed");
break;
};
},
packet = self.tun_task_response_rx.recv() => {
if let Some((_tag, packet)) = packet {
// Read recipient from env variable NYM_CLIENT_ADDR which is a base58
// string of the nym-address of the client that the packet should be
// sent back to.
//
// In the near future we will let the client expose it's nym-address
// directly, and after that, provide SURBS
let recipient = std::env::var("NYM_CLIENT_ADDR").ok().and_then(|addr| {
Recipient::try_from_base58_string(addr).ok()
});
if let Some(recipient) = recipient {
let lane = TransmissionLane::General;
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();
} else {
log::error!("NYM_CLIENT_ADDR not set or invalid");
}
} else {
log::trace!("IpPacketRouter [main loop]: stopping since channel closed");
break;
}
}
}
}
log::info!("IpPacketRouter: stopping");
Ok(())
}
async fn on_message(
&mut self,
reconstructed: ReconstructedMessage,
) -> Result<(), IpPacketRouterError> {
log::info!("Received message: {:?}", reconstructed.sender_tag);
let headers = etherparse::SlicedPacket::from_ip(&reconstructed.message).map_err(|err| {
log::warn!("Received non-IP packet: {err}");
IpPacketRouterError::PacketParseFailed { source: err }
})?;
let (src_addr, dst_addr): (IpAddr, IpAddr) = match headers.ip {
Some(etherparse::InternetSlice::Ipv4(ipv4_header, _)) => (
ipv4_header.source_addr().into(),
ipv4_header.destination_addr().into(),
),
Some(etherparse::InternetSlice::Ipv6(ipv6_header, _)) => (
ipv6_header.source_addr().into(),
ipv6_header.destination_addr().into(),
),
None => {
log::warn!("Received non-IP packet");
return Err(IpPacketRouterError::PacketMissingHeader);
}
};
log::info!("Received packet: {src_addr} -> {dst_addr}");
// 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();
Ok(())
}
}
@@ -128,6 +263,7 @@ impl IpForwarderBuilder {
// This is NOT in the SDK since we don't want to expose any of the client-core config types.
// We could however consider moving it to a crate in common in the future.
// TODO: refactor this function and its arguments
#[allow(unused)]
async fn create_mixnet_client(
config: &BaseClientConfig,
shutdown: TaskClient,
@@ -135,7 +271,7 @@ async fn create_mixnet_client(
custom_topology_provider: Option<Box<dyn TopologyProvider + Send + Sync>>,
wait_for_gateway: bool,
paths: &CommonClientPaths,
) -> Result<nym_sdk::mixnet::MixnetClient, IpForwarderError> {
) -> Result<nym_sdk::mixnet::MixnetClient, IpPacketRouterError> {
let debug_config = config.debug;
let storage_paths = nym_sdk::mixnet::StoragePaths::from(paths.clone());
@@ -143,7 +279,7 @@ async fn create_mixnet_client(
let mut client_builder =
nym_sdk::mixnet::MixnetClientBuilder::new_with_default_storage(storage_paths)
.await
.map_err(|err| IpForwarderError::FailedToSetupMixnetClient { source: err })?
.map_err(|err| IpPacketRouterError::FailedToSetupMixnetClient { source: err })?
.network_details(NymNetworkDetails::new_from_env())
.debug_config(debug_config)
.custom_shutdown(shutdown)
@@ -160,10 +296,10 @@ async fn create_mixnet_client(
let mixnet_client = client_builder
.build()
.map_err(|err| IpForwarderError::FailedToSetupMixnetClient { source: err })?;
.map_err(|err| IpPacketRouterError::FailedToSetupMixnetClient { source: err })?;
mixnet_client
.connect_to_mixnet()
.await
.map_err(|err| IpForwarderError::FailedToConnectToMixnet { source: err })
.map_err(|err| IpPacketRouterError::FailedToConnectToMixnet { source: err })
}