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

Merge pull request #4216 from nymtech/jon/per-client-mix-hops-in-ipr

Browse Source 
IPR: handle client specific number of mix hops
This commit is contained in:
Jon Häggblad
2023-12-05 21:11:18 +01:00
committed by GitHub
parent 8b5dc867cd ac14382a08
commit 755fd1d765
11 changed files with 572 additions and 399 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/client-core/src/client/inbound_messages.rs
+4 -4
View File
@@ -110,13 +110,13 @@ impl InputMessage {
data: Vec<u8>,
lane: TransmissionLane,
packet_type: Option<PacketType>,
mix_hops: u8,
mix_hops: Option<u8>,
) -> Self {
let message = InputMessage::Regular {
recipient,
data,
lane,
mix_hops: Some(mix_hops),
mix_hops,
};
if let Some(packet_type) = packet_type {
InputMessage::new_wrapper(message, packet_type)
@@ -154,14 +154,14 @@ impl InputMessage {
reply_surbs: u32,
lane: TransmissionLane,
packet_type: Option<PacketType>,
mix_hops: u8,
mix_hops: Option<u8>,
) -> Self {
let message = InputMessage::Anonymous {
recipient,
data,
reply_surbs,
lane,
mix_hops: Some(mix_hops),
mix_hops,
};
if let Some(packet_type) = packet_type {
InputMessage::new_wrapper(message, packet_type)
common/ip-packet-requests/src/lib.rs
+10
View File
@@ -108,16 +108,26 @@ pub enum IpPacketRequestData {
pub struct StaticConnectRequest {
pub request_id: u64,
pub ip: IpAddr,
// The nym-address the response should be sent back to
pub reply_to: Recipient,
// The number of mix node hops that responses should take, in addition to the entry and exit
// node. Zero means only client -> entry -> exit -> client.
pub reply_to_hops: Option<u8>,
// The average delay at each mix node, in milliseconds. Currently this is not supported by the
// ip packet router.
pub reply_to_avg_mix_delays: Option<f64>,
}
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
pub struct DynamicConnectRequest {
pub request_id: u64,
// The nym-address the response should be sent back to
pub reply_to: Recipient,
// The number of mix node hops that responses should take, in addition to the entry and exit
// node. Zero means only client -> entry -> exit -> client.
pub reply_to_hops: Option<u8>,
// The average delay at each mix node, in milliseconds. Currently this is not supported by the
// ip packet router.
pub reply_to_avg_mix_delays: Option<f64>,
}
gateway/src/node/mod.rs
+4 -4
View File
@@ -347,19 +347,19 @@ 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::IpPacketRouterBuilder::new(ip_opts.config.clone())
let mut ip_packet_router =
nym_ip_packet_router::IpPacketRouter::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)?
ip_packet_router = ip_packet_router.with_stored_topology(custom_mixnet)?
}
tokio::spawn(async move {
if let Err(err) = ip_builder.run_service_provider().await {
if let Err(err) = ip_packet_router.run_service_provider().await {
// no need to panic as we have passed a task client to the ip packet router so
// we're most likely already in the process of shutting down
error!("ip packet router has failed: {err}")
service-providers/ip-packet-router/src/error.rs
+14 -5
View File
@@ -33,6 +33,9 @@ pub enum IpPacketRouterError {
#[error("received packet has an invalid version: {0}")]
InvalidPacketVersion(u8),
#[error("failed to serialize response packet: {source}")]
FailedToSerializeResponsePacket { source: Box<bincode::ErrorKind> },
#[error("failed to deserialize tagged packet: {source}")]
FailedToDeserializeTaggedPacket { source: bincode::Error },
@@ -45,14 +48,12 @@ 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("failed to write packet to tun")]
FailedToWritePacketToTun,
#[error("failed to send packet to mixnet: {source}")]
FailedToSendPacketToMixnet { source: nym_sdk::Error },
#[error("the provided socket address, '{addr}' is not covered by the exit policy!")]
AddressNotCoveredByExitPolicy { addr: SocketAddr },
@@ -73,4 +74,12 @@ pub enum IpPacketRouterError {
#[error("can't setup an exit policy without any upstream urls")]
NoUpstreamExitPolicy,
#[error("no recipient in response packet")]
NoRecipientInResponse,
#[error("failed to update client activity")]
FailedToUpdateClientActivity,
}
pub type Result<T> = std::result::Result<T, IpPacketRouterError>;
service-providers/ip-packet-router/src/ip_packet_router.rs
+16 -338
View File
@@ -1,29 +1,19 @@
#![cfg_attr(not(target_os = "linux"), allow(dead_code))]
#![cfg_attr(not(target_os = "linux"), allow(unused_imports))]
use std::{collections::HashMap, net::IpAddr, path::Path};
use std::path::Path;
use futures::{channel::oneshot, StreamExt};
use futures::channel::oneshot;
use nym_client_core::{
client::mix_traffic::transceiver::GatewayTransceiver, HardcodedTopologyProvider,
TopologyProvider,
};
use nym_ip_packet_requests::{
DynamicConnectFailureReason, IpPacketRequest, IpPacketRequestData, IpPacketResponse,
StaticConnectFailureReason,
};
use nym_sdk::mixnet::{InputMessage, MixnetMessageSender, Recipient};
use nym_sphinx::receiver::ReconstructedMessage;
use nym_task::{connections::TransmissionLane, TaskClient, TaskHandle};
#[cfg(target_os = "linux")]
use tokio::io::AsyncWriteExt;
use nym_sdk::mixnet::Recipient;
use nym_task::{TaskClient, TaskHandle};
use crate::{
constants::{CLIENT_INACTIVITY_TIMEOUT, DISCONNECT_TIMER_INTERVAL},
error::IpPacketRouterError,
request_filter::{self, RequestFilter},
util::generate_new_ip,
util::parse_ip::{parse_packet, ParsedPacket},
Config,
};
@@ -43,7 +33,7 @@ impl OnStartData {
}
}
pub struct IpPacketRouterBuilder {
pub struct IpPacketRouter {
#[allow(unused)]
config: Config,
wait_for_gateway: bool,
@@ -53,7 +43,7 @@ pub struct IpPacketRouterBuilder {
on_start: Option<oneshot::Sender<OnStartData>>,
}
impl IpPacketRouterBuilder {
impl IpPacketRouter {
pub fn new(config: Config) -> Self {
Self {
config,
@@ -118,6 +108,8 @@ impl IpPacketRouterBuilder {
#[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).
use crate::{mixnet_listener, tun_listener};
let task_handle: TaskHandle = self.shutdown.map(Into::into).unwrap_or_default();
// Connect to the mixnet
@@ -144,28 +136,28 @@ impl IpPacketRouterBuilder {
// 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 (connected_clients, connected_clients_rx) = mixnet_listener::ConnectedClients::new();
// let (connected_client_tx, connected_client_rx) = tokio::sync::mpsc::unbounded_channel();
let tun_listener = crate::tun_listener::TunListener {
let tun_listener = 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,
connected_clients: connected_clients_rx,
};
tun_listener.start();
let request_filter = request_filter::RequestFilter::new(&self.config).await?;
request_filter.start_update_tasks().await;
let ip_packet_router_service = IpPacketRouter {
let mixnet_listener = mixnet_listener::MixnetListener {
_config: self.config,
request_filter: request_filter.clone(),
tun_writer,
mixnet_client,
task_handle,
connected_clients: Default::default(),
connected_client_tx,
connected_clients,
// connected_client_tx,
};
log::info!("The address of this client is: {self_address}");
@@ -181,320 +173,6 @@ impl IpPacketRouterBuilder {
}
}
ip_packet_router_service.run().await
mixnet_listener.run().await
}
}
#[cfg(target_os = "linux")]
struct IpPacketRouter {
_config: Config,
request_filter: request_filter::RequestFilter,
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>,
}
pub(crate) struct ConnectedClient {
pub(crate) nym_address: Recipient,
pub(crate) last_activity: std::time::Instant,
}
#[cfg(target_os = "linux")]
impl IpPacketRouter {
async fn on_static_connect_request(
&mut self,
connect_request: nym_ip_packet_requests::StaticConnectRequest,
) -> Result<Option<IpPacketResponse>, IpPacketRouterError> {
log::info!(
"Received static connect request from {sender_address}",
sender_address = connect_request.reply_to
);
let request_id = connect_request.request_id;
let requested_ip = connect_request.ip;
let reply_to = connect_request.reply_to;
// TODO: ignoring reply_to_hops and reply_to_avg_mix_delays for now
// Check that the IP is available in the set of connected clients
let is_ip_taken = self.connected_clients.contains_key(&requested_ip);
// Check that the nym address isn't already registered
let is_nym_address_taken = self
.connected_clients
.values()
.any(|client| client.nym_address == reply_to);
match (is_ip_taken, is_nym_address_taken) {
(true, true) => {
log::info!("Connecting an already connected client");
// Update the last activity time for the client
if let Some(client) = self.connected_clients.get_mut(&requested_ip) {
client.last_activity = std::time::Instant::now();
} else {
log::error!("Failed to update last activity time for client");
}
Ok(Some(IpPacketResponse::new_static_connect_success(
request_id, reply_to,
)))
}
(false, false) => {
log::info!("Connecting a new client");
self.connected_clients.insert(
requested_ip,
ConnectedClient {
nym_address: reply_to,
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,
)))
}
(true, false) => {
log::info!("Requested IP is not available");
Ok(Some(IpPacketResponse::new_static_connect_failure(
request_id,
reply_to,
StaticConnectFailureReason::RequestedIpAlreadyInUse,
)))
}
(false, true) => {
log::info!("Nym address is already registered");
Ok(Some(IpPacketResponse::new_static_connect_failure(
request_id,
reply_to,
StaticConnectFailureReason::RequestedNymAddressAlreadyInUse,
)))
}
}
}
async fn on_dynamic_connect_request(
&mut self,
connect_request: nym_ip_packet_requests::DynamicConnectRequest,
) -> Result<Option<IpPacketResponse>, IpPacketRouterError> {
log::info!(
"Received dynamic connect request from {sender_address}",
sender_address = connect_request.reply_to
);
let request_id = connect_request.request_id;
let reply_to = connect_request.reply_to;
// TODO: ignoring reply_to_hops and reply_to_avg_mix_delays for now
// Check if it's the same client connecting again, then we just reuse the same IP
// TODO: this is problematic. Until we sign connect requests this means you can spam people
// with return traffic
let existing_ip = self.connected_clients.iter().find_map(|(ip, client)| {
if client.nym_address == reply_to {
Some(*ip)
} else {
None
}
});
if let Some(existing_ip) = existing_ip {
log::info!("Found existing client for nym address");
// Update the last activity time for the client
if let Some(client) = self.connected_clients.get_mut(&existing_ip) {
client.last_activity = std::time::Instant::now();
} else {
log::error!("Failed to update last activity time for client");
}
return Ok(Some(IpPacketResponse::new_dynamic_connect_success(
request_id,
reply_to,
existing_ip,
)));
}
let Some(new_ip) = generate_new_ip::find_new_ip(&self.connected_clients) else {
log::info!("No available IP address");
return Ok(Some(IpPacketResponse::new_dynamic_connect_failure(
request_id,
reply_to,
DynamicConnectFailureReason::NoAvailableIp,
)));
};
self.connected_clients.insert(
new_ip,
ConnectedClient {
nym_address: reply_to,
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,
)))
}
async fn on_data_request(
&mut self,
data_request: nym_ip_packet_requests::DataRequest,
) -> Result<Option<IpPacketResponse>, IpPacketRouterError> {
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.
//
// For example, if we are running in a mode where we are only supposed to forward
// packets to a specific destination, we might want to forward them anyway.
//
// TODO: look into this
let ParsedPacket {
packet_type,
src_addr,
dst_addr,
dst,
} = parse_packet(&data_request.ip_packet)?;
let dst_str = dst.map_or(dst_addr.to_string(), |dst| dst.to_string());
log::info!("Received packet: {packet_type}: {src_addr} -> {dst_str}");
// Check if there is a connected client for this src_addr. If there is, update the last activity time
// for the client. If there isn't, drop the packet.
if let Some(client) = self.connected_clients.get_mut(&src_addr) {
client.last_activity = std::time::Instant::now();
} else {
log::info!("Dropping packet: no connected client for {src_addr}");
return Ok(None);
}
// Filter check
if let Some(dst) = dst {
if !self.request_filter.check_address(&dst).await {
log::warn!("Failed filter check: {dst}");
// TODO: we could consider sending back a response here
return Err(IpPacketRouterError::AddressFailedFilterCheck { addr: dst });
}
} else {
// TODO: we should also filter packets without port number
log::warn!("Ignoring filter check for packet without port number! TODO!");
}
// TODO: consider changing from Vec<u8> to bytes::Bytes?
let packet = data_request.ip_packet;
self.tun_writer
.write_all(&packet)
.await
.map_err(|_| IpPacketRouterError::FailedToWritePacketToTun)?;
Ok(None)
}
async fn on_reconstructed_message(
&mut self,
reconstructed: ReconstructedMessage,
) -> Result<Option<IpPacketResponse>, IpPacketRouterError> {
log::debug!(
"Received message with sender_tag: {:?}",
reconstructed.sender_tag
);
// Check version of request
if let Some(version) = reconstructed.message.first() {
// The idea is that in the future we can add logic here to parse older versions to stay
// backwards compatible.
if *version != nym_ip_packet_requests::CURRENT_VERSION {
log::warn!("Received packet with invalid version");
return Err(IpPacketRouterError::InvalidPacketVersion(*version));
}
}
let request = IpPacketRequest::from_reconstructed_message(&reconstructed)
.map_err(|err| IpPacketRouterError::FailedToDeserializeTaggedPacket { source: err })?;
match request.data {
IpPacketRequestData::StaticConnect(connect_request) => {
self.on_static_connect_request(connect_request).await
}
IpPacketRequestData::DynamicConnect(connect_request) => {
self.on_dynamic_connect_request(connect_request).await
}
IpPacketRequestData::Data(data_request) => self.on_data_request(data_request).await,
}
}
async fn run(mut self) -> Result<(), IpPacketRouterError> {
let mut task_client = self.task_handle.fork("main_loop");
let mut disconnect_timer = tokio::time::interval(DISCONNECT_TIMER_INTERVAL);
while !task_client.is_shutdown() {
tokio::select! {
_ = task_client.recv() => {
log::debug!("IpPacketRouter [main loop]: received shutdown");
},
_ = disconnect_timer.tick() => {
let now = std::time::Instant::now();
let inactive_clients: Vec<IpAddr> = self.connected_clients.iter()
.filter_map(|(ip, client)| {
if now.duration_since(client.last_activity) > CLIENT_INACTIVITY_TIMEOUT {
Some(*ip)
} else {
None
}
})
.collect();
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() => {
if let Some(msg) = msg {
match self.on_reconstructed_message(msg).await {
Ok(Some(response)) => {
let Some(recipient) = response.recipient() else {
log::error!("IpPacketRouter [main loop]: failed to get recipient from response");
continue;
};
let response_packet = response.to_bytes();
let Ok(response_packet) = response_packet else {
log::error!("Failed to serialize response packet");
continue;
};
let lane = TransmissionLane::General;
let packet_type = None;
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}");
};
},
Ok(None) => {
continue;
},
Err(err) => {
log::error!("Error handling mixnet message: {err}");
}
};
} else {
log::trace!("IpPacketRouter [main loop]: stopping since channel closed");
break;
};
},
}
}
log::debug!("IpPacketRouter: stopping");
Ok(())
}
}
pub(crate) enum ConnectedClientEvent {
Disconnect(IpAddr),
Connect(IpAddr, Box<Recipient>),
}
service-providers/ip-packet-router/src/lib.rs
+2 -1
View File
@@ -2,13 +2,14 @@
#![cfg_attr(not(target_os = "linux"), allow(unused_imports))]
pub use crate::config::Config;
pub use ip_packet_router::{IpPacketRouterBuilder, OnStartData};
pub use ip_packet_router::{IpPacketRouter, OnStartData};
pub mod config;
mod constants;
pub mod error;
mod ip_packet_router;
mod mixnet_client;
mod mixnet_listener;
mod request_filter;
mod tun_listener;
mod util;
service-providers/ip-packet-router/src/mixnet_listener.rs
+454
View File
@@ -0,0 +1,454 @@
use std::{collections::HashMap, net::IpAddr};
use futures::StreamExt;
use nym_ip_packet_requests::{
DynamicConnectFailureReason, IpPacketRequest, IpPacketRequestData, IpPacketResponse,
StaticConnectFailureReason,
};
use nym_sdk::mixnet::{MixnetMessageSender, Recipient};
use nym_sphinx::receiver::ReconstructedMessage;
use nym_task::TaskHandle;
#[cfg(target_os = "linux")]
use tokio::io::AsyncWriteExt;
use crate::{
constants::{CLIENT_INACTIVITY_TIMEOUT, DISCONNECT_TIMER_INTERVAL},
error::{IpPacketRouterError, Result},
request_filter::{self},
util::generate_new_ip,
util::{
create_message::create_input_message,
parse_ip::{parse_packet, ParsedPacket},
},
Config,
};
#[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>,
}
pub(crate) struct ConnectedClientsListener {
clients: HashMap<IpAddr, ConnectedClient>,
pub(crate) connected_client_rx: tokio::sync::mpsc::UnboundedReceiver<ConnectedClientEvent>,
}
impl ConnectedClientsListener {
pub(crate) fn get(&self, ip: &IpAddr) -> Option<&ConnectedClient> {
self.clients.get(ip)
}
pub(crate) fn update(&mut self, event: ConnectedClientEvent) {
match event {
ConnectedClientEvent::Connect(connected_event) => {
let ConnectEvent {
ip,
nym_address,
mix_hops,
} = *connected_event;
log::trace!("Connect client: {ip}");
self.clients.insert(
ip,
ConnectedClient {
nym_address,
mix_hops,
last_activity: std::time::Instant::now(),
},
);
}
ConnectedClientEvent::Disconnect(DisconnectEvent(ip)) => {
log::trace!("Disconnect client: {ip}");
self.clients.remove(&ip);
}
}
}
}
impl ConnectedClients {
pub(crate) fn new() -> (Self, ConnectedClientsListener) {
let (connected_client_tx, connected_client_rx) = tokio::sync::mpsc::unbounded_channel();
(
Self {
clients: Default::default(),
connected_client_tx,
},
ConnectedClientsListener {
clients: Default::default(),
connected_client_rx,
},
)
}
fn is_ip_connected(&self, ip: &IpAddr) -> bool {
self.clients.contains_key(ip)
}
fn is_nym_address_connected(&self, nym_address: &Recipient) -> bool {
self.clients
.values()
.any(|client| client.nym_address == *nym_address)
}
fn get_ip(&self, nym_address: &Recipient) -> Option<IpAddr> {
self.clients.iter().find_map(|(ip, client)| {
if client.nym_address == *nym_address {
Some(*ip)
} else {
None
}
})
}
fn get_client_by_nym_address(&self, nym_address: &Recipient) -> Option<&ConnectedClient> {
self.clients
.values()
.find(|client| client.nym_address == *nym_address)
}
fn connect(&mut self, ip: IpAddr, nym_address: Recipient, mix_hops: Option<u8>) {
self.clients.insert(
ip,
ConnectedClient {
nym_address,
mix_hops,
last_activity: std::time::Instant::now(),
},
);
self.connected_client_tx
.send(ConnectedClientEvent::Connect(Box::new(ConnectEvent {
ip,
nym_address,
mix_hops,
})))
.unwrap();
}
fn update_activity(&mut self, ip: &IpAddr) -> Result<()> {
if let Some(client) = self.clients.get_mut(ip) {
client.last_activity = std::time::Instant::now();
Ok(())
} else {
Err(IpPacketRouterError::FailedToUpdateClientActivity)
}
}
fn disconnect_inactive_clients(&mut self) {
let now = std::time::Instant::now();
let inactive_clients: Vec<IpAddr> = self
.clients
.iter()
.filter_map(|(ip, client)| {
if now.duration_since(client.last_activity) > CLIENT_INACTIVITY_TIMEOUT {
Some(*ip)
} else {
None
}
})
.collect();
for ip in inactive_clients {
log::info!("Disconnect inactive client: {ip}");
self.clients.remove(&ip);
self.connected_client_tx
.send(ConnectedClientEvent::Disconnect(DisconnectEvent(ip)))
.unwrap();
}
}
fn find_new_ip(&self) -> Option<IpAddr> {
generate_new_ip::find_new_ip(&self.clients)
}
}
pub(crate) struct ConnectedClient {
pub(crate) nym_address: Recipient,
pub(crate) mix_hops: Option<u8>,
pub(crate) last_activity: std::time::Instant,
}
#[cfg(target_os = "linux")]
impl MixnetListener {
async fn on_static_connect_request(
&mut self,
connect_request: nym_ip_packet_requests::StaticConnectRequest,
) -> Result<Option<IpPacketResponse>> {
log::info!(
"Received static connect request from {sender_address}",
sender_address = connect_request.reply_to
);
let request_id = connect_request.request_id;
let requested_ip = connect_request.ip;
let reply_to = connect_request.reply_to;
let reply_to_hops = connect_request.reply_to_hops;
// TODO: ignoring reply_to_avg_mix_delays for now
// Check that the IP is available in the set of connected clients
let is_ip_taken = self.connected_clients.is_ip_connected(&requested_ip);
// Check that the nym address isn't already registered
let is_nym_address_taken = self.connected_clients.is_nym_address_connected(&reply_to);
match (is_ip_taken, is_nym_address_taken) {
(true, true) => {
log::info!("Connecting an already connected client");
if self
.connected_clients
.update_activity(&requested_ip)
.is_err()
{
log::error!("Failed to update activity for client");
};
Ok(Some(IpPacketResponse::new_static_connect_success(
request_id, reply_to,
)))
}
(false, false) => {
log::info!("Connecting a new client");
self.connected_clients
.connect(requested_ip, reply_to, reply_to_hops);
Ok(Some(IpPacketResponse::new_static_connect_success(
request_id, reply_to,
)))
}
(true, false) => {
log::info!("Requested IP is not available");
Ok(Some(IpPacketResponse::new_static_connect_failure(
request_id,
reply_to,
StaticConnectFailureReason::RequestedIpAlreadyInUse,
)))
}
(false, true) => {
log::info!("Nym address is already registered");
Ok(Some(IpPacketResponse::new_static_connect_failure(
request_id,
reply_to,
StaticConnectFailureReason::RequestedNymAddressAlreadyInUse,
)))
}
}
}
async fn on_dynamic_connect_request(
&mut self,
connect_request: nym_ip_packet_requests::DynamicConnectRequest,
) -> Result<Option<IpPacketResponse>> {
log::info!(
"Received dynamic connect request from {sender_address}",
sender_address = connect_request.reply_to
);
let request_id = connect_request.request_id;
let reply_to = connect_request.reply_to;
let reply_to_hops = connect_request.reply_to_hops;
// TODO: ignoring reply_to_avg_mix_delays for now
// Check if it's the same client connecting again, then we just reuse the same IP
// TODO: this is problematic. Until we sign connect requests this means you can spam people
// with return traffic
if let Some(existing_ip) = self.connected_clients.get_ip(&reply_to) {
log::info!("Found existing client for nym address");
if self
.connected_clients
.update_activity(&existing_ip)
.is_err()
{
log::error!("Failed to update activity for client");
}
return Ok(Some(IpPacketResponse::new_dynamic_connect_success(
request_id,
reply_to,
existing_ip,
)));
}
let Some(new_ip) = self.connected_clients.find_new_ip() else {
log::info!("No available IP address");
return Ok(Some(IpPacketResponse::new_dynamic_connect_failure(
request_id,
reply_to,
DynamicConnectFailureReason::NoAvailableIp,
)));
};
self.connected_clients
.connect(new_ip, reply_to, reply_to_hops);
Ok(Some(IpPacketResponse::new_dynamic_connect_success(
request_id, reply_to, new_ip,
)))
}
async fn on_data_request(
&mut self,
data_request: nym_ip_packet_requests::DataRequest,
) -> Result<Option<IpPacketResponse>> {
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.
//
// For example, if we are running in a mode where we are only supposed to forward
// packets to a specific destination, we might want to forward them anyway.
//
// TODO: look into this
let ParsedPacket {
packet_type,
src_addr,
dst_addr,
dst,
} = parse_packet(&data_request.ip_packet)?;
let dst_str = dst.map_or(dst_addr.to_string(), |dst| dst.to_string());
log::info!("Received packet: {packet_type}: {src_addr} -> {dst_str}");
// Check if there is a connected client for this src_addr. If there is, update the last activity time
// for the client. If there isn't, drop the packet.
if self.connected_clients.update_activity(&src_addr).is_err() {
log::info!("Dropping packet: no connected client for {src_addr}");
return Ok(None);
}
// Filter check
if let Some(dst) = dst {
if !self.request_filter.check_address(&dst).await {
log::warn!("Failed filter check: {dst}");
// TODO: we could consider sending back a response here
return Err(IpPacketRouterError::AddressFailedFilterCheck { addr: dst });
}
} else {
// TODO: we should also filter packets without port number
log::warn!("Ignoring filter check for packet without port number! TODO!");
}
// TODO: consider changing from Vec<u8> to bytes::Bytes?
let packet = data_request.ip_packet;
self.tun_writer
.write_all(&packet)
.await
.map_err(|_| IpPacketRouterError::FailedToWritePacketToTun)?;
Ok(None)
}
async fn on_reconstructed_message(
&mut self,
reconstructed: ReconstructedMessage,
) -> Result<Option<IpPacketResponse>> {
log::debug!(
"Received message with sender_tag: {:?}",
reconstructed.sender_tag
);
// Check version of request
if let Some(version) = reconstructed.message.first() {
// The idea is that in the future we can add logic here to parse older versions to stay
// backwards compatible.
if *version != nym_ip_packet_requests::CURRENT_VERSION {
log::warn!("Received packet with invalid version");
return Err(IpPacketRouterError::InvalidPacketVersion(*version));
}
}
let request = IpPacketRequest::from_reconstructed_message(&reconstructed)
.map_err(|err| IpPacketRouterError::FailedToDeserializeTaggedPacket { source: err })?;
match request.data {
IpPacketRequestData::StaticConnect(connect_request) => {
self.on_static_connect_request(connect_request).await
}
IpPacketRequestData::DynamicConnect(connect_request) => {
self.on_dynamic_connect_request(connect_request).await
}
IpPacketRequestData::Data(data_request) => self.on_data_request(data_request).await,
}
}
// When an incoming mixnet message triggers a response that we send back, such as during
// connect handshake.
async fn handle_response(&self, response: IpPacketResponse) -> Result<()> {
let Some(recipient) = response.recipient() else {
log::error!("no recipient in response packet, this should NOT happen!");
return Err(IpPacketRouterError::NoRecipientInResponse);
};
let response_packet = response.to_bytes().map_err(|err| {
log::error!("Failed to serialize response packet");
IpPacketRouterError::FailedToSerializeResponsePacket { source: err }
})?;
// We could avoid this lookup if we check this when we create the response.
let mix_hops = self
.connected_clients
.get_client_by_nym_address(recipient)
.and_then(|c| c.mix_hops);
let input_message = create_input_message(*recipient, response_packet, mix_hops);
self.mixnet_client
.send(input_message)
.await
.map_err(|err| IpPacketRouterError::FailedToSendPacketToMixnet { source: err })
}
pub(crate) async fn run(mut self) -> Result<()> {
let mut task_client = self.task_handle.fork("main_loop");
let mut disconnect_timer = tokio::time::interval(DISCONNECT_TIMER_INTERVAL);
while !task_client.is_shutdown() {
tokio::select! {
_ = task_client.recv() => {
log::debug!("IpPacketRouter [main loop]: received shutdown");
},
_ = disconnect_timer.tick() => {
self.connected_clients.disconnect_inactive_clients();
},
msg = self.mixnet_client.next() => {
if let Some(msg) = msg {
match self.on_reconstructed_message(msg).await {
Ok(Some(response)) => {
if let Err(err) = self.handle_response(response).await {
log::error!("Mixnet listener failed to handle response: {err}");
}
},
Ok(None) => {
continue;
},
Err(err) => {
log::error!("Error handling mixnet message: {err}");
}
};
} else {
log::trace!("IpPacketRouter [main loop]: stopping since channel closed");
break;
};
},
}
}
log::debug!("IpPacketRouter: stopping");
Ok(())
}
}
pub(crate) enum ConnectedClientEvent {
Disconnect(DisconnectEvent),
Connect(Box<ConnectEvent>),
}
pub(crate) struct DisconnectEvent(pub(crate) IpAddr);
pub(crate) struct ConnectEvent {
pub(crate) ip: IpAddr,
pub(crate) nym_address: Recipient,
pub(crate) mix_hops: Option<u8>,
}
service-providers/ip-packet-router/src/tun_listener.rs
+48 -46
View File
@@ -1,13 +1,14 @@
use std::{collections::HashMap, net::IpAddr};
use nym_ip_packet_requests::IpPacketResponse;
use nym_sdk::mixnet::{InputMessage, MixnetMessageSender};
use nym_task::{connections::TransmissionLane, TaskClient};
use nym_sdk::mixnet::MixnetMessageSender;
use nym_task::TaskClient;
#[cfg(target_os = "linux")]
use tokio::io::AsyncReadExt;
use tokio::sync::mpsc::UnboundedReceiver;
use crate::{error::IpPacketRouterError, ip_packet_router, util::parse_ip::parse_dst_addr};
use crate::{
error::{IpPacketRouterError, Result},
mixnet_listener::{self},
util::{create_message::create_input_message, parse_ip::parse_dst_addr},
};
// Reads packet from TUN and writes to mixnet client
#[cfg(target_os = "linux")]
@@ -15,60 +16,61 @@ 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) task_client: TaskClient,
// A mirror of the one in IpPacketRouter
pub(crate) connected_clients: HashMap<IpAddr, ip_packet_router::ConnectedClient>,
pub(crate) connected_client_rx: UnboundedReceiver<ip_packet_router::ConnectedClientEvent>,
pub(crate) connected_clients: mixnet_listener::ConnectedClientsListener,
}
#[cfg(target_os = "linux")]
impl TunListener {
async fn run(mut self) -> Result<(), IpPacketRouterError> {
async fn handle_packet(&mut self, buf: &[u8], len: usize) -> Result<()> {
let Some(dst_addr) = parse_dst_addr(&buf[..len]) else {
log::warn!("Failed to parse packet");
return Ok(());
};
if let Some(mixnet_listener::ConnectedClient {
nym_address,
mix_hops,
..
}) = self.connected_clients.get(&dst_addr)
{
let packet = buf[..len].to_vec();
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");
}
Ok(())
}
async fn run(mut self) -> Result<()> {
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(ip_packet_router::ConnectedClientEvent::Connect(ip, nym_addr)) => {
log::trace!("Connect client: {ip}");
self.connected_clients.insert(ip, ip_packet_router::ConnectedClient {
nym_address: *nym_addr,
last_activity: std::time::Instant::now(),
});
// TODO: ConnectedClientsListener::update should poll the channel instead
event = self.connected_clients.connected_client_rx.recv() => match event {
Some(event) => self.connected_clients.update(event),
None => {
log::error!("TunListener: connected client channel closed");
break;
},
Some(ip_packet_router::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;
};
let recipient = self.connected_clients.get(&dst_addr).map(|c| c.nym_address);
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");
continue;
};
let input_message = InputMessage::new_regular(recipient, response_packet, lane, packet_type);
if let Err(err) = self.mixnet_client_sender.send(input_message).await {
log::error!("TunListener: failed to send packet to mixnet: {err}");
};
} else {
log::info!("No registered nym-address for packet - dropping");
if let Err(err) = self.handle_packet(&buf, len).await {
log::error!("tun: failed to handle packet: {err}");
}
},
Err(err) => {
service-providers/ip-packet-router/src/util/create_message.rs
+18
View File
@@ -0,0 +1,18 @@
use nym_sdk::mixnet::{InputMessage, Recipient};
use nym_task::connections::TransmissionLane;
pub(crate) fn create_input_message(
nym_address: Recipient,
response_packet: Vec<u8>,
mix_hops: Option<u8>,
) -> InputMessage {
let lane = TransmissionLane::General;
let packet_type = None;
InputMessage::new_regular_with_custom_hops(
nym_address,
response_packet,
lane,
packet_type,
mix_hops,
)
}
service-providers/ip-packet-router/src/util/generate_new_ip.rs
+1 -1
View File
@@ -3,7 +3,7 @@ use std::{
net::{IpAddr, Ipv4Addr},
};
use crate::{constants::TUN_DEVICE_ADDRESS, ip_packet_router::ConnectedClient};
use crate::{constants::TUN_DEVICE_ADDRESS, mixnet_listener::ConnectedClient};
// Find an available IP address in self.connected_clients
// TODO: make this nicer
service-providers/ip-packet-router/src/util/mod.rs
+1
View File
@@ -1,2 +1,3 @@
pub(crate) mod create_message;
pub(crate) mod generate_new_ip;
pub(crate) mod parse_ip;