diff --git a/Cargo.lock b/Cargo.lock index c545037965..0ecbfad25c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6608,6 +6608,7 @@ dependencies = [ "bincode", "bytes", "nym-sphinx", + "rand 0.8.5", "serde", ] diff --git a/common/ip-packet-requests/Cargo.toml b/common/ip-packet-requests/Cargo.toml index d77f0f9396..969152baf7 100644 --- a/common/ip-packet-requests/Cargo.toml +++ b/common/ip-packet-requests/Cargo.toml @@ -14,4 +14,5 @@ license.workspace = true bincode = "1.3.3" bytes = "1.5.0" nym-sphinx = { path = "../nymsphinx" } +rand = "0.8.5" serde = { workspace = true, features = ["derive"] } diff --git a/common/ip-packet-requests/src/lib.rs b/common/ip-packet-requests/src/lib.rs index ef7672bf17..95dc67076d 100644 --- a/common/ip-packet-requests/src/lib.rs +++ b/common/ip-packet-requests/src/lib.rs @@ -1,35 +1,270 @@ -#[derive(serde::Serialize, serde::Deserialize)] -pub struct TaggedIpPacket { - pub packet: bytes::Bytes, - pub return_address: nym_sphinx::addressing::clients::Recipient, - pub return_mix_hops: Option, - // pub return_mix_delays: Option, +use std::net::IpAddr; + +use nym_sphinx::addressing::clients::Recipient; +use serde::{Deserialize, Serialize}; + +pub const CURRENT_VERSION: u8 = 1; + +fn generate_random() -> u64 { + use rand::RngCore; + let mut rng = rand::rngs::OsRng; + rng.next_u64() } -impl TaggedIpPacket { - pub fn new( - packet: bytes::Bytes, - return_address: nym_sphinx::addressing::clients::Recipient, - return_mix_hops: Option, - ) -> Self { - TaggedIpPacket { - packet, - return_address, - return_mix_hops, +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct IpPacketRequest { + pub version: u8, + pub data: IpPacketRequestData, +} + +impl IpPacketRequest { + pub fn new_static_connect_request( + ip: IpAddr, + reply_to: Recipient, + reply_to_hops: Option, + reply_to_avg_mix_delays: Option, + ) -> (Self, u64) { + let request_id = generate_random(); + ( + Self { + version: CURRENT_VERSION, + data: IpPacketRequestData::StaticConnect(StaticConnectRequest { + request_id, + ip, + reply_to, + reply_to_hops, + reply_to_avg_mix_delays, + }), + }, + request_id, + ) + } + + pub fn new_dynamic_connect_request( + reply_to: Recipient, + reply_to_hops: Option, + reply_to_avg_mix_delays: Option, + ) -> (Self, u64) { + let request_id = generate_random(); + ( + Self { + version: CURRENT_VERSION, + data: IpPacketRequestData::DynamicConnect(DynamicConnectRequest { + request_id, + reply_to, + reply_to_hops, + reply_to_avg_mix_delays, + }), + }, + request_id, + ) + } + + pub fn new_ip_packet(ip_packet: bytes::Bytes) -> Self { + Self { + version: CURRENT_VERSION, + data: IpPacketRequestData::Data(DataRequest { ip_packet }), } } + pub fn id(&self) -> Option { + match &self.data { + IpPacketRequestData::StaticConnect(request) => Some(request.request_id), + IpPacketRequestData::DynamicConnect(request) => Some(request.request_id), + IpPacketRequestData::Data(_) => None, + } + } + + pub fn recipient(&self) -> Option<&Recipient> { + match &self.data { + IpPacketRequestData::StaticConnect(request) => Some(&request.reply_to), + IpPacketRequestData::DynamicConnect(request) => Some(&request.reply_to), + IpPacketRequestData::Data(_) => None, + } + } + + pub fn to_bytes(&self) -> Result, bincode::Error> { + use bincode::Options; + make_bincode_serializer().serialize(self) + } + pub fn from_reconstructed_message( message: &nym_sphinx::receiver::ReconstructedMessage, ) -> Result { use bincode::Options; make_bincode_serializer().deserialize(&message.message) } +} + +#[allow(clippy::large_enum_variant)] +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)] +pub enum IpPacketRequestData { + StaticConnect(StaticConnectRequest), + DynamicConnect(DynamicConnectRequest), + Data(DataRequest), +} + +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)] +pub struct StaticConnectRequest { + pub request_id: u64, + pub ip: IpAddr, + pub reply_to: Recipient, + pub reply_to_hops: Option, + pub reply_to_avg_mix_delays: Option, +} + +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)] +pub struct DynamicConnectRequest { + pub request_id: u64, + pub reply_to: Recipient, + pub reply_to_hops: Option, + pub reply_to_avg_mix_delays: Option, +} + +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)] +pub struct DataRequest { + pub ip_packet: bytes::Bytes, +} + +// --- + +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct IpPacketResponse { + pub version: u8, + pub data: IpPacketResponseData, +} + +impl IpPacketResponse { + pub fn new_static_connect_success(request_id: u64, reply_to: Recipient) -> Self { + Self { + version: CURRENT_VERSION, + data: IpPacketResponseData::StaticConnect(StaticConnectResponse { + request_id, + reply_to, + reply: StaticConnectResponseReply::Success, + }), + } + } + + pub fn new_static_connect_failure(request_id: u64, reply_to: Recipient) -> Self { + Self { + version: CURRENT_VERSION, + data: IpPacketResponseData::StaticConnect(StaticConnectResponse { + request_id, + reply_to, + reply: StaticConnectResponseReply::Failure, + }), + } + } + + pub fn new_dynamic_connect_success(request_id: u64, reply_to: Recipient, ip: IpAddr) -> Self { + Self { + version: CURRENT_VERSION, + data: IpPacketResponseData::DynamicConnect(DynamicConnectResponse { + request_id, + reply_to, + reply: DynamicConnectResponseReply::Success(DynamicConnectSuccess { ip }), + }), + } + } + + pub fn new_dynamic_connect_failure(request_id: u64, reply_to: Recipient) -> Self { + Self { + version: CURRENT_VERSION, + data: IpPacketResponseData::DynamicConnect(DynamicConnectResponse { + request_id, + reply_to, + reply: DynamicConnectResponseReply::Failure, + }), + } + } + + pub fn new_ip_packet(ip_packet: bytes::Bytes) -> Self { + Self { + version: CURRENT_VERSION, + data: IpPacketResponseData::Data(DataResponse { ip_packet }), + } + } + + pub fn id(&self) -> Option { + match &self.data { + IpPacketResponseData::StaticConnect(response) => Some(response.request_id), + IpPacketResponseData::DynamicConnect(response) => Some(response.request_id), + IpPacketResponseData::Data(_) => None, + } + } + + pub fn recipient(&self) -> Option<&Recipient> { + match &self.data { + IpPacketResponseData::StaticConnect(response) => Some(&response.reply_to), + IpPacketResponseData::DynamicConnect(response) => Some(&response.reply_to), + IpPacketResponseData::Data(_) => None, + } + } pub fn to_bytes(&self) -> Result, bincode::Error> { use bincode::Options; make_bincode_serializer().serialize(self) } + + pub fn from_reconstructed_message( + message: &nym_sphinx::receiver::ReconstructedMessage, + ) -> Result { + use bincode::Options; + make_bincode_serializer().deserialize(&message.message) + } +} + +#[allow(clippy::large_enum_variant)] +#[derive(Clone, Debug, Serialize, Deserialize)] +pub enum IpPacketResponseData { + StaticConnect(StaticConnectResponse), + DynamicConnect(DynamicConnectResponse), + Data(DataResponse), +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct StaticConnectResponse { + pub request_id: u64, + pub reply_to: Recipient, + pub reply: StaticConnectResponseReply, +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +pub enum StaticConnectResponseReply { + Success, + Failure, +} +impl StaticConnectResponseReply { + pub fn is_success(&self) -> bool { + match self { + StaticConnectResponseReply::Success => true, + StaticConnectResponseReply::Failure => false, + } + } +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct DynamicConnectResponse { + pub request_id: u64, + pub reply_to: Recipient, + pub reply: DynamicConnectResponseReply, +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +pub enum DynamicConnectResponseReply { + Success(DynamicConnectSuccess), + Failure, +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct DynamicConnectSuccess { + pub ip: IpAddr, +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct DataResponse { + pub ip_packet: bytes::Bytes, } fn make_bincode_serializer() -> impl bincode::Options { @@ -38,3 +273,63 @@ fn make_bincode_serializer() -> impl bincode::Options { .with_big_endian() .with_varint_encoding() } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn check_size_of_request() { + let connect = IpPacketRequest { + version: 4, + data: IpPacketRequestData::StaticConnect( + StaticConnectRequest { + request_id: 123, + ip: IpAddr::from([10, 0, 0, 1]), + reply_to: Recipient::try_from_base58_string("D1rrpsysCGCYXy9saP8y3kmNpGtJZUXN9SvFoUcqAsM9.9Ssso1ea5NfkbMASdiseDSjTN1fSWda5SgEVjdSN4CvV@GJqd3ZxpXWSNxTfx7B1pPtswpetH4LnJdFeLeuY5KUuN").unwrap(), + reply_to_hops: None, + reply_to_avg_mix_delays: None, + }, + ) + }; + assert_eq!(connect.to_bytes().unwrap().len(), 107); + } + + #[test] + fn check_size_of_data() { + let data = IpPacketRequest { + version: 4, + data: IpPacketRequestData::Data(DataRequest { + ip_packet: bytes::Bytes::from(vec![1u8; 32]), + }), + }; + assert_eq!(data.to_bytes().unwrap().len(), 35); + } + + #[test] + fn serialize_and_deserialize_data_request() { + let data = IpPacketRequest { + version: 4, + data: IpPacketRequestData::Data(DataRequest { + ip_packet: bytes::Bytes::from(vec![1, 2, 4, 2, 5]), + }), + }; + + let serialized = data.to_bytes().unwrap(); + let deserialized = IpPacketRequest::from_reconstructed_message( + &nym_sphinx::receiver::ReconstructedMessage { + message: serialized, + sender_tag: None, + }, + ) + .unwrap(); + + assert_eq!(deserialized.version, 4); + assert_eq!( + deserialized.data, + IpPacketRequestData::Data(DataRequest { + ip_packet: bytes::Bytes::from(vec![1, 2, 4, 2, 5]), + }) + ); + } +} diff --git a/common/tun/src/linux/tun_device.rs b/common/tun/src/linux/tun_device.rs index 84a62a1220..709f5fec3b 100644 --- a/common/tun/src/linux/tun_device.rs +++ b/common/tun/src/linux/tun_device.rs @@ -198,7 +198,7 @@ impl TunDevice { async fn handle_tun_read(&self, packet: &[u8]) -> Result<(), TunDeviceError> { let ParsedAddresses { src_addr, dst_addr } = parse_src_dst_address(packet)?; log::debug!( - "iface: read Packet({src_addr} -> {dst_addr}, {} bytes)", + "iface: read Packet({dst_addr} <- {src_addr}, {} bytes)", packet.len(), ); diff --git a/service-providers/ip-packet-router/src/error.rs b/service-providers/ip-packet-router/src/error.rs index 9c417a8f6e..797dc78b81 100644 --- a/service-providers/ip-packet-router/src/error.rs +++ b/service-providers/ip-packet-router/src/error.rs @@ -30,6 +30,9 @@ pub enum IpPacketRouterError { #[error("the entity wrapping the network requester has disconnected")] DisconnectedParent, + #[error("received packet has an invalid version: {0}")] + InvalidPacketVersion(u8), + #[error("failed to deserialize tagged packet: {source}")] FailedToDeserializeTaggedPacket { source: bincode::Error }, diff --git a/service-providers/ip-packet-router/src/lib.rs b/service-providers/ip-packet-router/src/lib.rs index d77fcba90b..d85b1a744c 100644 --- a/service-providers/ip-packet-router/src/lib.rs +++ b/service-providers/ip-packet-router/src/lib.rs @@ -11,6 +11,7 @@ use nym_client_core::{ client::mix_traffic::transceiver::GatewayTransceiver, config::disk_persistence::CommonClientPaths, HardcodedTopologyProvider, TopologyProvider, }; +use nym_ip_packet_requests::{IpPacketRequest, IpPacketRequestData, IpPacketResponse}; use nym_sdk::{ mixnet::{InputMessage, MixnetMessageSender, Recipient}, NymNetworkDetails, @@ -189,8 +190,116 @@ struct IpPacketRouter { task_handle: TaskHandle, } -#[allow(unused)] +#[cfg_attr(not(target_os = "linux"), allow(unused))] impl IpPacketRouter { + async fn on_static_connect_request( + &mut self, + connect_request: nym_ip_packet_requests::StaticConnectRequest, + ) -> Result, 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 + + Ok(Some(IpPacketResponse::new_static_connect_failure( + request_id, reply_to, + ))) + } + + async fn on_dynamic_connect_request( + &mut self, + connect_request: nym_ip_packet_requests::DynamicConnectRequest, + ) -> Result, IpPacketRouterError> { + log::info!( + "Received dynamic connect request from {sender_address}", + sender_address = connect_request.reply_to + ); + log::info!("Dropping request: dynamic connect requests are not yet supported"); + Ok(None) + } + + async fn on_data_request( + &mut self, + data_request: nym_ip_packet_requests::DataRequest, + ) -> Result, IpPacketRouterError> { + log::info!("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}"); + + // 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: set the tag correctly. Can we just reuse sender_tag? + // TODO: consider changing from Vec to bytes::Bytes? + let peer_tag = 0; + self.tun_task_tx + .try_send((peer_tag, data_request.ip_packet.into())) + .map_err(|err| IpPacketRouterError::FailedToSendPacketToTun { source: err })?; + + Ok(None) + } + async fn on_reconstructed_message( + &mut self, + reconstructed: ReconstructedMessage, + ) -> Result, 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"); @@ -201,8 +310,31 @@ impl IpPacketRouter { }, msg = self.mixnet_client.next() => { if let Some(msg) = msg { - if let Err(err) = self.on_message(msg).await { - log::error!("Error handling mixnet message: {err}"); + 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"); @@ -224,7 +356,12 @@ impl IpPacketRouter { if let Some(recipient) = recipient { let lane = TransmissionLane::General; let packet_type = None; - let input_message = InputMessage::new_regular(recipient, packet, lane, packet_type); + 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.send(input_message).await { log::error!("IpPacketRouter [main loop]: failed to send packet to mixnet: {err}"); @@ -243,59 +380,6 @@ impl IpPacketRouter { log::info!("IpPacketRouter: stopping"); Ok(()) } - - async fn on_message( - &mut self, - reconstructed: ReconstructedMessage, - ) -> Result<(), IpPacketRouterError> { - log::debug!( - "Received message with sender_tag: {:?}", - reconstructed.sender_tag - ); - - let tagged_packet = - nym_ip_packet_requests::TaggedIpPacket::from_reconstructed_message(&reconstructed) - .map_err(|err| IpPacketRouterError::FailedToDeserializeTaggedPacket { - source: err, - })?; - - // 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(&tagged_packet.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}"); - - // 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: set the tag correctly. Can we just reuse sender_tag? - let peer_tag = 0; - self.tun_task_tx - .try_send((peer_tag, tagged_packet.packet.into())) - .map_err(|err| IpPacketRouterError::FailedToSendPacketToTun { source: err })?; - - Ok(()) - } } struct ParsedPacket<'a> {