Implement ip-packet-router connect request response outline (#4177)

* Try revised request response connect types * wip: try to get data requests work * Also serialize response * tweak log * Add code path for connect response fail * Convenience functions on ip packet request * Remove commented out code * Restore log to debug
2023-11-27 12:17:58 +01:00
parent 39d714f2c0
commit 44395f8466
6 changed files with 458 additions and 74 deletions
@@ -6608,6 +6608,7 @@ dependencies = [
 "bincode",
 "bytes",
 "nym-sphinx",
+ "rand 0.8.5",
 "serde",
 ]

@@ -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"] }
@@ -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<u8>,
-    // pub return_mix_delays: Option<f64>,
+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<u8>,
-    ) -> 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<u8>,
+        reply_to_avg_mix_delays: Option<f64>,
+    ) -> (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<u8>,
+        reply_to_avg_mix_delays: Option<f64>,
+    ) -> (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<u64> {
+        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<Vec<u8>, bincode::Error> {
+        use bincode::Options;
+        make_bincode_serializer().serialize(self)
+    }
+
    pub fn from_reconstructed_message(
        message: &nym_sphinx::receiver::ReconstructedMessage,
    ) -> Result<Self, bincode::Error> {
        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<u8>,
+    pub reply_to_avg_mix_delays: Option<f64>,
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
+pub struct DynamicConnectRequest {
+    pub request_id: u64,
+    pub reply_to: Recipient,
+    pub reply_to_hops: Option<u8>,
+    pub reply_to_avg_mix_delays: Option<f64>,
+}
+
+#[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<u64> {
+        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<Vec<u8>, bincode::Error> {
        use bincode::Options;
        make_bincode_serializer().serialize(self)
    }
+
+    pub fn from_reconstructed_message(
+        message: &nym_sphinx::receiver::ReconstructedMessage,
+    ) -> Result<Self, bincode::Error> {
+        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]),
+            })
+        );
+    }
+}
@@ -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(),
        );

@@ -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 },

@@ -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<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
+
+        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<Option<IpPacketResponse>, 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<Option<IpPacketResponse>, 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<u8> 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<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");

@@ -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> {