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

Add IPv6 support to IPPR (#4431)

Browse Source
This commit is contained in:
Bogdan-Ștefan Neacşu
2024-03-01 19:28:21 +02:00
committed by GitHub
parent 9a6f96b5e0
commit aebd386382
10 changed files with 293 additions and 154 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/ip-packet-requests/src/lib.rs
+12 -1
View File
@@ -1,8 +1,19 @@
use serde::{Deserialize, Serialize};
use std::net::{Ipv4Addr, Ipv6Addr};
pub mod codec;
pub mod request;
pub mod response;
pub const CURRENT_VERSION: u8 = 3;
// version 3: initial version
// version 4: IPv6 support
pub const CURRENT_VERSION: u8 = 4;
#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub struct IPPair {
pub ipv4: Ipv4Addr,
pub ipv6: Ipv6Addr,
}
fn make_bincode_serializer() -> impl bincode::Options {
use bincode::Options;
common/ip-packet-requests/src/request.rs
+19 -8
View File
@@ -1,9 +1,15 @@
use std::net::IpAddr;
use std::fmt::{Display, Formatter};
use nym_sphinx::addressing::clients::Recipient;
use serde::{Deserialize, Serialize};
use crate::{make_bincode_serializer, CURRENT_VERSION};
use crate::{make_bincode_serializer, IPPair, CURRENT_VERSION};
impl Display for IPPair {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
writeln!(f, "IPv4: {}, IPV6: {}", self.ipv4, self.ipv6)
}
}
fn generate_random() -> u64 {
use rand::RngCore;
@@ -19,7 +25,7 @@ pub struct IpPacketRequest {
impl IpPacketRequest {
pub fn new_static_connect_request(
ip: IpAddr,
ips: IPPair,
reply_to: Recipient,
reply_to_hops: Option<u8>,
reply_to_avg_mix_delays: Option<f64>,
@@ -31,7 +37,7 @@ impl IpPacketRequest {
version: CURRENT_VERSION,
data: IpPacketRequestData::StaticConnect(StaticConnectRequest {
request_id,
ip,
ips,
reply_to,
reply_to_hops,
reply_to_avg_mix_delays,
@@ -137,7 +143,7 @@ pub enum IpPacketRequestData {
pub struct StaticConnectRequest {
pub request_id: u64,
pub ip: IpAddr,
pub ips: IPPair,
// The nym-address the response should be sent back to
pub reply_to: Recipient,
@@ -210,6 +216,8 @@ pub struct HealthRequest {
#[cfg(test)]
mod tests {
use super::*;
use std::net::{Ipv4Addr, Ipv6Addr};
use std::str::FromStr;
#[test]
fn check_size_of_request() {
@@ -218,15 +226,18 @@ mod tests {
data: IpPacketRequestData::StaticConnect(
StaticConnectRequest {
request_id: 123,
ip: IpAddr::from([10, 0, 0, 1]),
ips: IPPair {
ipv4: Ipv4Addr::from_str("10.0.0.1").unwrap(),
ipv6: Ipv6Addr::from_str("2001:db8:a160::1").unwrap(),
},
reply_to: Recipient::try_from_base58_string("D1rrpsysCGCYXy9saP8y3kmNpGtJZUXN9SvFoUcqAsM9.9Ssso1ea5NfkbMASdiseDSjTN1fSWda5SgEVjdSN4CvV@GJqd3ZxpXWSNxTfx7B1pPtswpetH4LnJdFeLeuY5KUuN").unwrap(),
reply_to_hops: None,
reply_to_avg_mix_delays: None,
buffer_timeout: None,
},
)
),
};
assert_eq!(connect.to_bytes().unwrap().len(), 108);
assert_eq!(connect.to_bytes().unwrap().len(), 123);
}
#[test]
common/ip-packet-requests/src/response.rs
+4 -6
View File
@@ -1,9 +1,7 @@
use std::net::IpAddr;
use nym_sphinx::addressing::clients::Recipient;
use serde::{Deserialize, Serialize};
use crate::{make_bincode_serializer, CURRENT_VERSION};
use crate::{make_bincode_serializer, IPPair, CURRENT_VERSION};
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct IpPacketResponse {
@@ -38,13 +36,13 @@ impl IpPacketResponse {
}
}
pub fn new_dynamic_connect_success(request_id: u64, reply_to: Recipient, ip: IpAddr) -> Self {
pub fn new_dynamic_connect_success(request_id: u64, reply_to: Recipient, ips: IPPair) -> Self {
Self {
version: CURRENT_VERSION,
data: IpPacketResponseData::DynamicConnect(DynamicConnectResponse {
request_id,
reply_to,
reply: DynamicConnectResponseReply::Success(DynamicConnectSuccess { ip }),
reply: DynamicConnectResponseReply::Success(DynamicConnectSuccess { ips }),
}),
}
}
@@ -263,7 +261,7 @@ impl DynamicConnectResponseReply {
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct DynamicConnectSuccess {
pub ip: IpAddr,
pub ips: IPPair,
}
#[derive(Clone, Debug, Serialize, Deserialize, thiserror::Error)]
common/tun/src/linux/tun_device.rs
+38 -14
View File
@@ -1,3 +1,4 @@
use std::net::Ipv6Addr;
use std::{
collections::HashMap,
net::{IpAddr, Ipv4Addr},
@@ -19,6 +20,12 @@ const TUN_WRITE_TIMEOUT_MS: u64 = 1000;
#[derive(thiserror::Error, Debug)]
pub enum TunDeviceError {
#[error("{0}")]
IO(#[from] std::io::Error),
#[error("{0}")]
TokioTun(#[from] tokio_tun::Error),
#[error("timeout writing to tun device, dropping packet")]
TunWriteTimeout,
@@ -44,14 +51,18 @@ pub enum TunDeviceError {
FailedToLockPeer,
}
fn setup_tokio_tun_device(name: &str, address: Ipv4Addr, netmask: Ipv4Addr) -> tokio_tun::Tun {
fn setup_tokio_tun_device(
name: &str,
address: Ipv4Addr,
netmask: Ipv4Addr,
) -> Result<tokio_tun::Tun, TunDeviceError> {
log::info!("Creating TUN device with: address={address}, netmask={netmask}");
// Read MTU size from env variable NYM_MTU_SIZE, else default to 1420.
let mtu = std::env::var("NYM_MTU_SIZE")
.map(|mtu| mtu.parse().expect("NYM_MTU_SIZE must be a valid integer"))
.unwrap_or(1420);
log::info!("Using MTU size: {mtu}");
tokio_tun::Tun::builder()
Ok(tokio_tun::Tun::builder()
.name(name)
.tap(false)
.packet_info(false)
@@ -59,8 +70,7 @@ fn setup_tokio_tun_device(name: &str, address: Ipv4Addr, netmask: Ipv4Addr) -> t
.up()
.address(address)
.netmask(netmask)
.try_build()
.expect("Failed to setup tun device, do you have permission?")
.try_build()?)
}
pub struct TunDevice {
@@ -103,16 +113,18 @@ pub struct NatInner {
pub struct TunDeviceConfig {
pub base_name: String,
pub ip: Ipv4Addr,
pub netmask: Ipv4Addr,
pub ipv4: Ipv4Addr,
pub netmaskv4: Ipv4Addr,
pub ipv6: Ipv6Addr,
pub netmaskv6: String,
}
impl TunDevice {
pub fn new(
routing_mode: RoutingMode,
config: TunDeviceConfig,
) -> (Self, TunTaskTx, TunTaskResponseRx) {
let tun = Self::new_device_only(config);
) -> Result<(Self, TunTaskTx, TunTaskResponseRx), TunDeviceError> {
let tun = Self::new_device_only(config)?;
// Channels to communicate with the other tasks
let (tun_task_tx, tun_task_rx) = tun_task_channel();
@@ -125,20 +137,32 @@ impl TunDevice {
routing_mode,
};
(tun_device, tun_task_tx, tun_task_response_rx)
Ok((tun_device, tun_task_tx, tun_task_response_rx))
}
pub fn new_device_only(config: TunDeviceConfig) -> tokio_tun::Tun {
pub fn new_device_only(config: TunDeviceConfig) -> Result<tokio_tun::Tun, TunDeviceError> {
let TunDeviceConfig {
base_name,
ip,
netmask,
ipv4,
netmaskv4,
ipv6,
netmaskv6,
} = config;
let name = format!("{base_name}%d");
let tun = setup_tokio_tun_device(&name, ip, netmask);
let tun = setup_tokio_tun_device(&name, ipv4, netmaskv4)?;
log::info!("Created TUN device: {}", tun.name());
tun
std::process::Command::new("ip")
.args([
"-6",
"addr",
"add",
&format!("{}/{}", ipv6, netmaskv6),
"dev",
&tun.name(),
])
.output()?;
Ok(tun)
}
// Send outbound packets out on the wild internet
service-providers/ip-packet-router/src/constants.rs
+6 -2
View File
@@ -1,9 +1,13 @@
use std::net::{Ipv4Addr, Ipv6Addr};
use std::time::Duration;
// The interface used to route traffic
pub const TUN_BASE_NAME: &str = "nymtun";
pub const TUN_DEVICE_ADDRESS: &str = "10.0.0.1";
pub const TUN_DEVICE_NETMASK: &str = "255.255.255.0";
pub const TUN_DEVICE_ADDRESS_V4: Ipv4Addr = Ipv4Addr::new(10, 0, 0, 1);
pub const TUN_DEVICE_NETMASK_V4: Ipv4Addr = Ipv4Addr::new(255, 255, 255, 0);
pub const TUN_DEVICE_ADDRESS_V6: Ipv6Addr = Ipv6Addr::new(0x2001, 0xdb8, 0xa160, 0, 0, 0, 0, 0x1); // 2001:db8:a160::1
pub const TUN_DEVICE_NETMASK_V6: &str = "120";
// We routinely check if any clients needs to be disconnected at this interval
pub(crate) const DISCONNECT_TIMER_INTERVAL: Duration = Duration::from_secs(10);
service-providers/ip-packet-router/src/error.rs
+4
View File
@@ -11,6 +11,10 @@ pub enum IpPacketRouterError {
#[error("client-core error: {0}")]
ClientCoreError(#[from] ClientCoreError),
#[cfg(target_os = "linux")]
#[error("tun device error: {0}")]
TunDeviceError(#[from] nym_tun::tun_device::TunDeviceError),
#[error("failed to load configuration file: {0}")]
FailedToLoadConfig(String),
service-providers/ip-packet-router/src/ip_packet_router.rs
+5 -3
View File
@@ -133,11 +133,13 @@ impl IpPacketRouter {
// Create the TUN device that we interact with the rest of the world with
let config = nym_tun::tun_device::TunDeviceConfig {
base_name: crate::constants::TUN_BASE_NAME.to_string(),
ip: crate::constants::TUN_DEVICE_ADDRESS.parse().unwrap(),
netmask: crate::constants::TUN_DEVICE_NETMASK.parse().unwrap(),
ipv4: crate::constants::TUN_DEVICE_ADDRESS_V4,
netmaskv4: crate::constants::TUN_DEVICE_NETMASK_V4,
ipv6: crate::constants::TUN_DEVICE_ADDRESS_V6,
netmaskv6: crate::constants::TUN_DEVICE_NETMASK_V6.to_string(),
};
let (tun_reader, tun_writer) =
tokio::io::split(nym_tun::tun_device::TunDevice::new_device_only(config));
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
service-providers/ip-packet-router/src/mixnet_listener.rs
+140 -90
View File
@@ -1,7 +1,6 @@
use std::{
collections::HashMap,
net::{IpAddr, SocketAddr},
};
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
use std::sync::Arc;
use std::{collections::HashMap, net::SocketAddr};
use bytes::{Bytes, BytesMut};
use futures::StreamExt;
@@ -12,6 +11,7 @@ use nym_ip_packet_requests::{
DynamicConnectFailureReason, ErrorResponseReply, IpPacketResponse,
StaticConnectFailureReason,
},
IPPair,
};
use nym_sdk::mixnet::{MixnetMessageSender, Recipient};
use nym_sphinx::receiver::ReconstructedMessage;
@@ -19,6 +19,7 @@ use nym_task::TaskHandle;
use tap::TapFallible;
#[cfg(target_os = "linux")]
use tokio::io::AsyncWriteExt;
use tokio::sync::RwLock;
use tokio_util::codec::Decoder;
use crate::{
@@ -37,7 +38,8 @@ use crate::{
pub(crate) struct ConnectedClients {
// The set of connected clients
clients: HashMap<IpAddr, ConnectedClient>,
clients_ipv4_mapping: HashMap<Ipv4Addr, ConnectedClient>,
clients_ipv6_mapping: HashMap<Ipv6Addr, ConnectedClient>,
// Notify the tun listener when a new client connects or disconnects
tun_listener_connected_client_tx: tokio::sync::mpsc::UnboundedSender<ConnectedClientEvent>,
@@ -48,46 +50,62 @@ impl ConnectedClients {
let (connected_client_tx, connected_client_rx) = tokio::sync::mpsc::unbounded_channel();
(
Self {
clients: Default::default(),
clients_ipv4_mapping: Default::default(),
clients_ipv6_mapping: Default::default(),
tun_listener_connected_client_tx: connected_client_tx,
},
tun_listener::ConnectedClientsListener::new(connected_client_rx),
)
}
fn is_ip_connected(&self, ip: &IpAddr) -> bool {
self.clients.contains_key(ip)
fn is_ip_connected(&self, ips: &IPPair) -> bool {
self.clients_ipv4_mapping.contains_key(&ips.ipv4)
|| self.clients_ipv6_mapping.contains_key(&ips.ipv6)
}
fn get_client_from_ip_mut(&mut self, ip: &IpAddr) -> Option<&mut ConnectedClient> {
self.clients.get_mut(ip)
match ip {
IpAddr::V4(ip) => self.clients_ipv4_mapping.get_mut(ip),
IpAddr::V6(ip) => self.clients_ipv6_mapping.get_mut(ip),
}
}
fn is_nym_address_connected(&self, nym_address: &Recipient) -> bool {
self.clients
self.clients_ipv4_mapping
.values()
.any(|client| client.nym_address == *nym_address)
}
fn lookup_ip_from_nym_address(&self, nym_address: &Recipient) -> Option<IpAddr> {
self.clients.iter().find_map(|(ip, client)| {
if client.nym_address == *nym_address {
Some(*ip)
} else {
None
}
})
fn lookup_ip_from_nym_address(&self, nym_address: &Recipient) -> Option<IPPair> {
self.clients_ipv4_mapping
.iter()
.find_map(|(ipv4, connected_client)| {
if connected_client.nym_address == *nym_address {
Some(IPPair {
ipv4: *ipv4,
ipv6: connected_client.ipv6,
})
} else {
None
}
})
}
fn lookup_client_from_nym_address(&self, nym_address: &Recipient) -> Option<&ConnectedClient> {
self.clients
.values()
.find(|client| client.nym_address == *nym_address)
self.clients_ipv4_mapping
.iter()
.find_map(|(_, connected_client)| {
if connected_client.nym_address == *nym_address {
Some(connected_client)
} else {
None
}
})
}
fn connect(
&mut self,
ip: IpAddr,
ips: IPPair,
nym_address: Recipient,
mix_hops: Option<u8>,
forward_from_tun_tx: tokio::sync::mpsc::UnboundedSender<Vec<u8>>,
@@ -96,21 +114,25 @@ impl ConnectedClients {
) {
// 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.
self.clients.insert(
ip,
ConnectedClient {
let client = ConnectedClient {
nym_address,
ipv6: ips.ipv6,
mix_hops,
last_activity: Arc::new(RwLock::new(std::time::Instant::now())),
_close_tx: Arc::new(CloseTx {
nym_address,
mix_hops,
last_activity: std::time::Instant::now(),
close_tx: Some(close_tx),
handle,
},
);
inner: Some(close_tx),
}),
handle: Arc::new(handle),
};
log::info!("Inserting {} and {}", ips.ipv4, ips.ipv6);
self.clients_ipv4_mapping.insert(ips.ipv4, client.clone());
self.clients_ipv6_mapping.insert(ips.ipv6, client);
// Send the connected client info to the tun listener, which will use it to forward packets
// to the connected client handler.
self.tun_listener_connected_client_tx
.send(ConnectedClientEvent::Connect(Box::new(ConnectEvent {
ip,
ips,
forward_from_tun_tx,
})))
.tap_err(|err| {
@@ -119,9 +141,9 @@ impl ConnectedClients {
.ok();
}
fn update_activity(&mut self, ip: &IpAddr) -> Result<()> {
if let Some(client) = self.clients.get_mut(ip) {
client.last_activity = std::time::Instant::now();
async fn update_activity(&mut self, ips: &IPPair) -> Result<()> {
if let Some(client) = self.clients_ipv4_mapping.get(&ips.ipv4) {
*client.last_activity.write().await = std::time::Instant::now();
Ok(())
} else {
Err(IpPacketRouterError::FailedToUpdateClientActivity)
@@ -129,12 +151,18 @@ impl ConnectedClients {
}
// Identify connected client handlers that have stopped without being told to stop
fn get_finished_client_handlers(&mut self) -> Vec<(IpAddr, Recipient)> {
self.clients
fn get_finished_client_handlers(&mut self) -> Vec<(IPPair, Recipient)> {
self.clients_ipv4_mapping
.iter_mut()
.filter_map(|(ip, client)| {
if client.handle.is_finished() {
Some((*ip, client.nym_address))
.filter_map(|(ip, connected_client)| {
if connected_client.handle.is_finished() {
Some((
IPPair {
ipv4: *ip,
ipv6: connected_client.ipv6,
},
connected_client.nym_address,
))
} else {
None
}
@@ -142,26 +170,32 @@ impl ConnectedClients {
.collect()
}
fn get_inactive_clients(&mut self) -> Vec<(IpAddr, Recipient)> {
async fn get_inactive_clients(&mut self) -> Vec<(IPPair, Recipient)> {
let now = std::time::Instant::now();
self.clients
.iter()
.filter_map(|(ip, client)| {
if now.duration_since(client.last_activity) > CLIENT_MIXNET_INACTIVITY_TIMEOUT {
Some((*ip, client.nym_address))
} else {
None
}
})
.collect()
let mut ret = vec![];
for (ip, connected_client) in self.clients_ipv4_mapping.iter() {
if now.duration_since(*connected_client.last_activity.read().await)
> CLIENT_MIXNET_INACTIVITY_TIMEOUT
{
ret.push((
IPPair {
ipv4: *ip,
ipv6: connected_client.ipv6,
},
connected_client.nym_address,
))
}
}
ret
}
fn disconnect_stopped_client_handlers(&mut self, stopped_clients: Vec<(IpAddr, Recipient)>) {
for (ip, _) in &stopped_clients {
log::info!("Disconnect stopped client: {ip}");
self.clients.remove(ip);
fn disconnect_stopped_client_handlers(&mut self, stopped_clients: Vec<(IPPair, Recipient)>) {
for (ips, _) in &stopped_clients {
log::info!("Disconnect stopped client: {ips}");
self.clients_ipv4_mapping.remove(&ips.ipv4);
self.clients_ipv6_mapping.remove(&ips.ipv6);
self.tun_listener_connected_client_tx
.send(ConnectedClientEvent::Disconnect(DisconnectEvent(*ip)))
.send(ConnectedClientEvent::Disconnect(DisconnectEvent(*ips)))
.tap_err(|err| {
log::error!("Failed to send disconnect event: {err}");
})
@@ -169,12 +203,13 @@ impl ConnectedClients {
}
}
fn disconnect_inactive_clients(&mut self, inactive_clients: Vec<(IpAddr, Recipient)>) {
for (ip, _) in &inactive_clients {
log::info!("Disconnect inactive client: {ip}");
self.clients.remove(ip);
fn disconnect_inactive_clients(&mut self, inactive_clients: Vec<(IPPair, Recipient)>) {
for (ips, _) in &inactive_clients {
log::info!("Disconnect inactive client: {ips}");
self.clients_ipv4_mapping.remove(&ips.ipv4);
self.clients_ipv6_mapping.remove(&ips.ipv6);
self.tun_listener_connected_client_tx
.send(ConnectedClientEvent::Disconnect(DisconnectEvent(*ip)))
.send(ConnectedClientEvent::Disconnect(DisconnectEvent(*ips)))
.tap_err(|err| {
log::error!("Failed to send disconnect event: {err}");
})
@@ -182,40 +217,49 @@ impl ConnectedClients {
}
}
fn find_new_ip(&self) -> Option<IpAddr> {
generate_new_ip::find_new_ip(&self.clients)
fn find_new_ip(&self) -> Option<IPPair> {
generate_new_ip::find_new_ips(&self.clients_ipv4_mapping, &self.clients_ipv6_mapping)
}
}
pub(crate) struct CloseTx {
pub(crate) nym_address: Recipient,
// 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) inner: Option<tokio::sync::oneshot::Sender<()>>,
}
#[derive(Clone)]
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,
// The assigned IPv6 address of this client
pub(crate) ipv6: Ipv6Addr,
// 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,
pub(crate) last_activity: Arc<RwLock<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<()>>,
pub(crate) _close_tx: Arc<CloseTx>,
// Handle for the connected client handler
pub(crate) handle: tokio::task::JoinHandle<()>,
pub(crate) handle: Arc<tokio::task::JoinHandle<()>>,
}
impl ConnectedClient {
fn update_activity(&mut self) {
self.last_activity = std::time::Instant::now();
async fn update_activity(&self) {
*self.last_activity.write().await = std::time::Instant::now();
}
}
impl Drop for ConnectedClient {
impl Drop for CloseTx {
fn drop(&mut self) {
log::debug!("signal to close client: {}", self.nym_address);
if let Some(close_tx) = self.close_tx.take() {
if let Some(close_tx) = self.inner.take() {
close_tx.send(()).ok();
}
}
@@ -259,7 +303,7 @@ impl MixnetListener {
);
let request_id = connect_request.request_id;
let requested_ip = connect_request.ip;
let requested_ips = connect_request.ips;
let reply_to = connect_request.reply_to;
let reply_to_hops = connect_request.reply_to_hops;
// TODO: add to connect request
@@ -267,7 +311,7 @@ impl MixnetListener {
// 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);
let is_ip_taken = self.connected_clients.is_ip_connected(&requested_ips);
// Check that the nym address isn't already registered
let is_nym_address_taken = self.connected_clients.is_nym_address_connected(&reply_to);
@@ -277,7 +321,8 @@ impl MixnetListener {
log::info!("Connecting an already connected client");
if self
.connected_clients
.update_activity(&requested_ip)
.update_activity(&requested_ips)
.await
.is_err()
{
log::error!("Failed to update activity for client");
@@ -300,7 +345,7 @@ impl MixnetListener {
// Register the new client in the set of connected clients
self.connected_clients.connect(
requested_ip,
requested_ips,
reply_to,
reply_to_hops,
forward_from_tun_tx,
@@ -350,11 +395,12 @@ impl MixnetListener {
// 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.lookup_ip_from_nym_address(&reply_to) {
if let Some(existing_ips) = self.connected_clients.lookup_ip_from_nym_address(&reply_to) {
log::info!("Found existing client for nym address");
if self
.connected_clients
.update_activity(&existing_ip)
.update_activity(&existing_ips)
.await
.is_err()
{
log::error!("Failed to update activity for client");
@@ -362,11 +408,11 @@ impl MixnetListener {
return Ok(Some(IpPacketResponse::new_dynamic_connect_success(
request_id,
reply_to,
existing_ip,
existing_ips,
)));
}
let Some(new_ip) = self.connected_clients.find_new_ip() else {
let Some(new_ips) = self.connected_clients.find_new_ip() else {
log::info!("No available IP address");
return Ok(Some(IpPacketResponse::new_dynamic_connect_failure(
request_id,
@@ -386,7 +432,7 @@ impl MixnetListener {
// Register the new client in the set of connected clients
self.connected_clients.connect(
new_ip,
new_ips,
reply_to,
reply_to_hops,
forward_from_tun_tx,
@@ -394,7 +440,7 @@ impl MixnetListener {
handle,
);
Ok(Some(IpPacketResponse::new_dynamic_connect_success(
request_id, reply_to, new_ip,
request_id, reply_to, new_ips,
)))
}
@@ -428,7 +474,7 @@ impl MixnetListener {
if let Some(connected_client) = self.connected_clients.get_client_from_ip_mut(&src_addr) {
// Keep track of activity so we can disconnect inactive clients
connected_client.update_activity();
connected_client.update_activity().await;
// For packets without a port, use 0.
let dst = dst.unwrap_or_else(|| SocketAddr::new(dst_addr, 0));
@@ -539,9 +585,9 @@ impl MixnetListener {
}
}
fn handle_disconnect_timer(&mut self) {
async fn handle_disconnect_timer(&mut self) {
let stopped_clients = self.connected_clients.get_finished_client_handlers();
let inactive_clients = self.connected_clients.get_inactive_clients();
let inactive_clients = self.connected_clients.get_inactive_clients().await;
// TODO: Send disconnect responses to all disconnected clients
//for (ip, nym_address) in stopped_clients.iter().chain(disconnected_clients.iter()) {
@@ -571,10 +617,14 @@ impl MixnetListener {
})?;
// We could avoid this lookup if we check this when we create the response.
let mix_hops = self
let mix_hops = if let Some(c) = self
.connected_clients
.lookup_client_from_nym_address(recipient)
.and_then(|c| c.mix_hops);
{
c.mix_hops
} else {
None
};
let input_message = create_input_message(*recipient, response_packet, mix_hops);
self.mixnet_client
@@ -613,7 +663,7 @@ impl MixnetListener {
log::debug!("IpPacketRouter [main loop]: received shutdown");
},
_ = disconnect_timer.tick() => {
self.handle_disconnect_timer();
self.handle_disconnect_timer().await;
},
msg = self.mixnet_client.next() => {
if let Some(msg) = msg {
@@ -642,9 +692,9 @@ pub(crate) enum ConnectedClientEvent {
Connect(Box<ConnectEvent>),
}
pub(crate) struct DisconnectEvent(pub(crate) IpAddr);
pub(crate) struct DisconnectEvent(pub(crate) IPPair);
pub(crate) struct ConnectEvent {
pub(crate) ip: IpAddr,
pub(crate) ips: IPPair,
pub(crate) forward_from_tun_tx: tokio::sync::mpsc::UnboundedSender<Vec<u8>>,
}
service-providers/ip-packet-router/src/tun_listener.rs
+30 -12
View File
@@ -1,5 +1,7 @@
use std::{collections::HashMap, net::IpAddr};
use std::collections::HashMap;
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
use nym_ip_packet_requests::IPPair;
use nym_task::TaskClient;
#[cfg(target_os = "linux")]
use tokio::io::AsyncReadExt;
@@ -15,10 +17,12 @@ use crate::{
// It's even ok if this is slightly out of date
pub(crate) struct ConnectedClientMirror {
pub(crate) forward_from_tun_tx: tokio::sync::mpsc::UnboundedSender<Vec<u8>>,
pub(crate) ips: IPPair,
}
pub(crate) struct ConnectedClientsListener {
clients: HashMap<IpAddr, ConnectedClientMirror>,
clients_ipv4: HashMap<Ipv4Addr, ConnectedClientMirror>,
clients_ipv6: HashMap<Ipv6Addr, ConnectedClientMirror>,
connected_client_rx:
tokio::sync::mpsc::UnboundedReceiver<mixnet_listener::ConnectedClientEvent>,
}
@@ -30,35 +34,48 @@ impl ConnectedClientsListener {
>,
) -> Self {
ConnectedClientsListener {
clients: HashMap::new(),
clients_ipv4: HashMap::new(),
clients_ipv6: HashMap::new(),
connected_client_rx,
}
}
pub(crate) fn get(&self, ip: &IpAddr) -> Option<&ConnectedClientMirror> {
self.clients.get(ip)
match ip {
IpAddr::V4(ip) => self.clients_ipv4.get(ip),
IpAddr::V6(ip) => self.clients_ipv6.get(ip),
}
}
pub(crate) fn update(&mut self, event: mixnet_listener::ConnectedClientEvent) {
match event {
mixnet_listener::ConnectedClientEvent::Connect(connected_event) => {
let mixnet_listener::ConnectEvent {
ip,
ips,
forward_from_tun_tx,
} = *connected_event;
log::trace!("Connect client: {ip}");
self.clients.insert(
ip,
log::trace!("Connect client: {ips}");
self.clients_ipv4.insert(
ips.ipv4,
ConnectedClientMirror {
forward_from_tun_tx: forward_from_tun_tx.clone(),
ips,
},
);
self.clients_ipv6.insert(
ips.ipv6,
ConnectedClientMirror {
forward_from_tun_tx,
ips,
},
);
}
mixnet_listener::ConnectedClientEvent::Disconnect(
mixnet_listener::DisconnectEvent(ip),
mixnet_listener::DisconnectEvent(ips),
) => {
log::trace!("Disconnect client: {ip}");
self.clients.remove(&ip);
log::trace!("Disconnect client: {ips}");
self.clients_ipv4.remove(&ips.ipv4);
self.clients_ipv6.remove(&ips.ipv6);
}
}
}
@@ -82,6 +99,7 @@ impl TunListener {
if let Some(ConnectedClientMirror {
forward_from_tun_tx,
ips,
}) = self.connected_clients.get(&dst_addr)
{
let packet = buf[..len].to_vec();
@@ -89,7 +107,7 @@ impl TunListener {
log::warn!("Failed to forward packet to connected client {dst_addr}: disconnecting it from tun listener");
self.connected_clients
.update(mixnet_listener::ConnectedClientEvent::Disconnect(
mixnet_listener::DisconnectEvent(dst_addr),
mixnet_listener::DisconnectEvent(*ips),
));
}
} else {
service-providers/ip-packet-router/src/util/generate_new_ip.rs
+35 -18
View File
@@ -1,38 +1,55 @@
use std::{
collections::HashMap,
net::{IpAddr, Ipv4Addr},
};
use nym_ip_packet_requests::IPPair;
use std::net::Ipv6Addr;
use std::{collections::HashMap, net::Ipv4Addr};
use crate::{constants::TUN_DEVICE_ADDRESS, mixnet_listener::ConnectedClient};
use crate::constants::{TUN_DEVICE_ADDRESS_V4, TUN_DEVICE_ADDRESS_V6};
// Find an available IP address in self.connected_clients
// TODO: make this nicer
fn generate_random_ip_within_subnet() -> Ipv4Addr {
fn generate_random_ips_within_subnet() -> IPPair {
let mut rng = rand::thread_rng();
// Generate a random number in the range 1-254
let last_octet = rand::Rng::gen_range(&mut rng, 1..=254);
Ipv4Addr::new(10, 0, 0, last_octet)
let ipv4 = Ipv4Addr::new(10, 0, 0, last_octet);
let ipv6 = Ipv6Addr::new(0x2001, 0x0db8, 0xa160, 0, 0, 0, 0, last_octet as u16);
IPPair { ipv4, ipv6 }
}
fn is_ip_taken(
connected_clients: &HashMap<IpAddr, ConnectedClient>,
tun_ip: Ipv4Addr,
ip: Ipv4Addr,
fn is_ip_taken<T>(
connected_clients_ipv4: &HashMap<Ipv4Addr, T>,
connected_clients_ipv6: &HashMap<Ipv6Addr, T>,
tun_ips: IPPair,
ips: IPPair,
) -> bool {
connected_clients.contains_key(&ip.into()) || ip == tun_ip
connected_clients_ipv4.contains_key(&ips.ipv4)
|| connected_clients_ipv6.contains_key(&ips.ipv6)
|| ips.ipv4 == tun_ips.ipv4
|| ips.ipv6 == tun_ips.ipv6
}
// TODO: brute force approach. We could consider using a more efficient algorithm
pub(crate) fn find_new_ip(connected_clients: &HashMap<IpAddr, ConnectedClient>) -> Option<IpAddr> {
let mut new_ip = generate_random_ip_within_subnet();
pub(crate) fn find_new_ips<T>(
connected_clients_ipv4: &HashMap<Ipv4Addr, T>,
connected_clients_ipv6: &HashMap<Ipv6Addr, T>,
) -> Option<IPPair> {
let mut new_ips = generate_random_ips_within_subnet();
let mut tries = 0;
let tun_ip = TUN_DEVICE_ADDRESS.parse::<Ipv4Addr>().unwrap();
while is_ip_taken(connected_clients, tun_ip, new_ip) {
new_ip = generate_random_ip_within_subnet();
let tun_ips = IPPair {
ipv4: TUN_DEVICE_ADDRESS_V4,
ipv6: TUN_DEVICE_ADDRESS_V6,
};
while is_ip_taken(
connected_clients_ipv4,
connected_clients_ipv6,
tun_ips,
new_ips,
) {
new_ips = generate_random_ips_within_subnet();
tries += 1;
if tries > 100 {
return None;
}
}
Some(new_ip.into())
Some(new_ips)
}