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

Deduplicate and clean up

Browse Source
This commit is contained in:
Jon Häggblad
2024-02-08 11:58:27 +01:00
parent 14bf5645b1
commit d5e332ad39
5 changed files with 102 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/ip-packet-requests/src/request.rs
+7
View File
@@ -121,6 +121,8 @@ pub enum IpPacketRequestData {
Data(DataRequest),
}
// A static connect request is when the client provides the internal IP address it will use on the
// ip packet router.
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
pub struct StaticConnectRequest {
pub request_id: u64,
@@ -135,6 +137,8 @@ pub struct StaticConnectRequest {
pub reply_to_avg_mix_delays: Option<f64>,
}
// A dynamic connect request is when the client does not provide the internal IP address it will use
// on the ip packet router, and instead requests one to be assigned to it.
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
pub struct DynamicConnectRequest {
pub request_id: u64,
@@ -148,6 +152,8 @@ pub struct DynamicConnectRequest {
pub reply_to_avg_mix_delays: Option<f64>,
}
// A disconnect request is when the client wants to disconnect from the ip packet router and free
// up the allocated IP address.
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
pub struct DisconnectRequest {
pub request_id: u64,
@@ -155,6 +161,7 @@ pub struct DisconnectRequest {
pub reply_to: Recipient,
}
// A data request is when the client wants to send an IP packet to a destination.
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
pub struct DataRequest {
pub ip_packet: bytes::Bytes,
service-providers/ip-packet-router/src/connected_client_handler.rs
+32 -31
View File
@@ -1,15 +1,12 @@
use std::{collections::HashMap, net::IpAddr};
// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use nym_ip_packet_requests::response::IpPacketResponse;
use nym_sdk::mixnet::{MixnetMessageSender, Recipient};
use nym_task::TaskClient;
#[cfg(target_os = "linux")]
use tokio::io::AsyncReadExt;
use crate::{
error::{IpPacketRouterError, Result},
mixnet_listener::{self},
util::{create_message::create_input_message, parse_ip::parse_dst_addr},
util::create_message::create_input_message,
};
// Data flow
@@ -21,26 +18,38 @@ use crate::{
pub(crate) struct ConnectedClientHandler {
nym_address: Recipient,
mix_hops: Option<u8>,
tun_rx: tokio::sync::mpsc::UnboundedReceiver<Vec<u8>>,
forward_from_tun_rx: tokio::sync::mpsc::UnboundedReceiver<Vec<u8>>,
mixnet_client_sender: nym_sdk::mixnet::MixnetClientSender,
close_rx: tokio::sync::oneshot::Receiver<()>,
}
impl ConnectedClientHandler {
pub(crate) fn new(
nym_address: Recipient,
mix_hops: Option<u8>,
tun_rx: tokio::sync::mpsc::UnboundedReceiver<Vec<u8>>,
pub(crate) fn launch(
reply_to: Recipient,
reply_to_hops: Option<u8>,
mixnet_client_sender: nym_sdk::mixnet::MixnetClientSender,
close_rx: tokio::sync::oneshot::Receiver<()>,
) -> Self {
ConnectedClientHandler {
nym_address,
mix_hops,
tun_rx,
) -> (
tokio::sync::mpsc::UnboundedSender<Vec<u8>>,
tokio::sync::oneshot::Sender<()>,
) {
let (close_tx, close_rx) = tokio::sync::oneshot::channel();
let (forward_from_tun_tx, forward_from_tun_rx) = tokio::sync::mpsc::unbounded_channel();
let connected_client_handler = ConnectedClientHandler {
nym_address: reply_to,
mix_hops: reply_to_hops,
forward_from_tun_rx,
mixnet_client_sender,
close_rx,
}
};
tokio::spawn(async move {
if let Err(err) = connected_client_handler.run().await {
log::error!("connected client handler has failed: {err}")
}
});
(forward_from_tun_tx, close_tx)
}
async fn handle_packet(&mut self, packet: Vec<u8>) -> Result<()> {
@@ -52,41 +61,33 @@ impl ConnectedClientHandler {
self.mixnet_client_sender
.send(input_message)
.await
.map_err(|err| IpPacketRouterError::FailedToSendPacketToMixnet { source: err })?;
Ok(())
.map_err(|err| IpPacketRouterError::FailedToSendPacketToMixnet { source: err })
}
async fn run(mut self) -> Result<()> {
loop {
tokio::select! {
_ = &mut self.close_rx => {
// WIP(JON): downgrade to trace once confirmed to work
log::warn!("ConnectedClientHandler: received shutdown");
break;
},
packet = self.tun_rx.recv() => match packet {
packet = self.forward_from_tun_rx.recv() => match packet {
Some(packet) => {
if let Err(err) = self.handle_packet(packet).await {
log::error!("connected client handler: failed to handle packet: {err}");
}
},
None => {
log::error!("connected client handler: tun channel closed");
log::debug!("connected client handler: tun channel closed");
break;
}
}
}
}
// WIP(JON): downgrade to debug once confirmed to work
log::warn!("ConnectedClientHandler: exiting");
Ok(())
}
pub(crate) fn start(self) {
tokio::spawn(async move {
if let Err(err) = self.run().await {
log::error!("connected client handler has failed: {err}")
}
});
}
}
service-providers/ip-packet-router/src/ip_packet_router.rs
-3
View File
@@ -142,11 +142,9 @@ impl IpPacketRouter {
// Channel used by the IpPacketRouter to signal connected and disconnected clients to the
// TunListener
let (connected_clients, connected_clients_rx) = mixnet_listener::ConnectedClients::new();
// let (connected_client_tx, connected_client_rx) = tokio::sync::mpsc::unbounded_channel();
let tun_listener = tun_listener::TunListener {
tun_reader,
mixnet_client_sender: mixnet_client.split_sender(),
task_client: task_handle.get_handle(),
connected_clients: connected_clients_rx,
};
@@ -162,7 +160,6 @@ impl IpPacketRouter {
mixnet_client,
task_handle,
connected_clients,
// connected_client_tx,
};
log::info!("The address of this client is: {self_address}");
service-providers/ip-packet-router/src/mixnet_listener.rs
+59 -44
View File
@@ -31,16 +31,6 @@ use crate::{
},
};
#[cfg(target_os = "linux")]
pub(crate) struct MixnetListener {
pub(crate) _config: Config,
pub(crate) request_filter: request_filter::RequestFilter,
pub(crate) tun_writer: tokio::io::WriteHalf<tokio_tun::Tun>,
pub(crate) mixnet_client: nym_sdk::mixnet::MixnetClient,
pub(crate) task_handle: TaskHandle,
pub(crate) connected_clients: ConnectedClients,
}
pub(crate) struct ConnectedClients {
clients: HashMap<IpAddr, ConnectedClient>,
connected_client_tx: tokio::sync::mpsc::UnboundedSender<ConnectedClientEvent>,
@@ -156,22 +146,19 @@ impl ConnectedClients {
}
pub(crate) struct ConnectedClient {
// The nym address of the connected client that we are communicating with on the other side of
// the mixnet
pub(crate) nym_address: Recipient,
pub(crate) mix_hops: Option<u8>,
pub(crate) last_activity: std::time::Instant,
// Send to connected clients listener to stop
// This is inside an Option only because we want to send in Drop
pub(crate) close_tx: Option<tokio::sync::oneshot::Sender<()>>,
}
impl Drop for ConnectedClient {
fn drop(&mut self) {
log::info!("Dropping connected client: {}", self.nym_address);
if let Some(close_tx) = self.close_tx.take() {
log::info!("Sending close signal to connected client handler");
close_tx.send(()).unwrap();
}
}
// Number of mix node hops that the client has requested to use
pub(crate) mix_hops: Option<u8>,
// Keep track of last activity so we can disconnect inactive clients
pub(crate) last_activity: std::time::Instant,
// Send to connected clients listener to stop. This is option only because we need to take
// ownership of it when the client is dropped.
pub(crate) close_tx: Option<tokio::sync::oneshot::Sender<()>>,
}
impl ConnectedClient {
@@ -180,8 +167,44 @@ impl ConnectedClient {
}
}
impl Drop for ConnectedClient {
fn drop(&mut self) {
// WIP(JON): downgrade to trace once confirmed to work
log::info!("Dropping connected client: {}", self.nym_address);
if let Some(close_tx) = self.close_tx.take() {
// WIP(JON): downgrade to trace once confirmed to work
log::info!("Sending close signal to connected client handler");
close_tx.send(()).unwrap();
}
}
}
#[cfg(target_os = "linux")]
pub(crate) struct MixnetListener {
// The configuration for the mixnet listener
pub(crate) _config: Config,
// The request filter that we use to check if a packet should be forwarded
pub(crate) request_filter: request_filter::RequestFilter,
// The TUN device that we use to send and receive packets from the internet
pub(crate) tun_writer: tokio::io::WriteHalf<tokio_tun::Tun>,
// The mixnet client that we use to send and receive packets from the mixnet
pub(crate) mixnet_client: nym_sdk::mixnet::MixnetClient,
// The task handle for the main loop
pub(crate) task_handle: TaskHandle,
// The map of connected clients that the mixnet listener keeps track of. It monitors
// activity and disconnects clients that have been inactive for too long.
pub(crate) connected_clients: ConnectedClients,
}
#[cfg(target_os = "linux")]
impl MixnetListener {
// Receving a static connect request from a client with an IP provided that we assign to them,
// if it's available. If it's not available, we send a failure response.
async fn on_static_connect_request(
&mut self,
connect_request: nym_ip_packet_requests::request::StaticConnectRequest,
@@ -220,20 +243,15 @@ impl MixnetListener {
(false, false) => {
log::info!("Connecting a new client");
// Start the ConnectedClientHandler for the new client
let (close_tx, close_rx) = tokio::sync::oneshot::channel();
let (forward_from_tun_tx, forward_from_tun_rx) =
tokio::sync::mpsc::unbounded_channel();
let connected_client_handler =
connected_client_handler::ConnectedClientHandler::new(
// Spawn the ConnectedClientHandler for the new client
let (forward_from_tun_tx, close_tx) =
connected_client_handler::ConnectedClientHandler::launch(
reply_to,
reply_to_hops,
forward_from_tun_rx,
self.mixnet_client.split_sender(),
close_rx,
);
connected_client_handler.start();
// Register the new client in the set of connected clients
self.connected_clients.connect(
requested_ip,
reply_to,
@@ -307,18 +325,15 @@ impl MixnetListener {
)));
};
// Start the ConnectedClientHandler for the new client
let (close_tx, close_rx) = tokio::sync::oneshot::channel();
let (forward_from_tun_tx, forward_from_tun_rx) = tokio::sync::mpsc::unbounded_channel();
let connected_client_handler = connected_client_handler::ConnectedClientHandler::new(
reply_to,
reply_to_hops,
forward_from_tun_rx,
self.mixnet_client.split_sender(),
close_rx,
);
connected_client_handler.start();
// Spawn the ConnectedClientHandler for the new client
let (forward_from_tun_tx, close_tx) =
connected_client_handler::ConnectedClientHandler::launch(
reply_to,
reply_to_hops,
self.mixnet_client.split_sender(),
);
// Register the new client in the set of connected clients
self.connected_clients.connect(
new_ip,
reply_to,
service-providers/ip-packet-router/src/tun_listener.rs
+4 -17
View File
@@ -1,15 +1,14 @@
use std::{collections::HashMap, net::IpAddr};
use nym_ip_packet_requests::response::IpPacketResponse;
use nym_sdk::mixnet::{MixnetMessageSender, Recipient};
use nym_sdk::mixnet::Recipient;
use nym_task::TaskClient;
#[cfg(target_os = "linux")]
use tokio::io::AsyncReadExt;
use crate::{
error::{IpPacketRouterError, Result},
error::Result,
mixnet_listener::{self},
util::{create_message::create_input_message, parse_ip::parse_dst_addr},
util::parse_ip::parse_dst_addr,
};
pub(crate) struct ConnectedClientMirror {
@@ -76,7 +75,7 @@ impl ConnectedClientsListener {
#[cfg(target_os = "linux")]
pub(crate) struct TunListener {
pub(crate) tun_reader: tokio::io::ReadHalf<tokio_tun::Tun>,
pub(crate) mixnet_client_sender: nym_sdk::mixnet::MixnetClientSender,
// pub(crate) mixnet_client_sender: nym_sdk::mixnet::MixnetClientSender,
pub(crate) task_client: TaskClient,
pub(crate) connected_clients: ConnectedClientsListener,
}
@@ -98,18 +97,6 @@ impl TunListener {
{
let packet = buf[..len].to_vec();
forward_from_tun_tx.send(packet).unwrap();
// let response_packet = IpPacketResponse::new_ip_packet(packet.into())
// .to_bytes()
// .map_err(|err| IpPacketRouterError::FailedToSerializeResponsePacket {
// source: err,
// })?;
// let input_message = create_input_message(*nym_address, response_packet, *mix_hops);
//
// self.mixnet_client_sender
// .send(input_message)
// .await
// .map_err(|err| IpPacketRouterError::FailedToSendPacketToMixnet { source: err })?;
} else {
log::info!("No registered nym-address for packet - dropping");
}