Outfox integration (#3331)

* Experiment with serde * Framed encoding serde POC * Outfox framing * Outfox rest compat (#3333) * Outfox forwarding compat * Tidy up interface * PacketSize compat * Address PR comments * Rebase on develop commit 342883fcbe Author: durch <durch@users.noreply.github.com> Date: Thu Apr 27 09:17:18 2023 +0200 Put back PacketType 1 commit 61a0ee5a19 Author: Tommy Verrall <tommyvez@protonmail.com> Date: Wed Apr 26 16:37:29 2023 +0100 change output for cpu-cycle management logs commit 3956109c7e Author: Tommy Verrall <tommy@nymtech.net> Date: Wed Apr 26 12:13:22 2023 +0100 change the workflow file to build with cpucycles commit 8d725b13c5 Author: durch <durch@users.noreply.github.com> Date: Mon Apr 24 13:14:58 2023 +0200 Outfox client compat commit 4d166c389b Author: durch <durch@users.noreply.github.com> Date: Fri Apr 21 00:30:46 2023 +0200 Address PR comments commit 145c3c1223 Author: durch <durch@users.noreply.github.com> Date: Fri Apr 21 00:12:35 2023 +0200 Rename PacketMode commit cbd654d6fd Author: Drazen Urch <drazen@urch.eu> Date: Thu Apr 20 23:59:40 2023 +0200 Outfox rest compat (#3333) * Outfox forwarding compat * Tidy up interface * PacketSize compat commit e7be91a94c Author: durch <durch@users.noreply.github.com> Date: Wed Apr 19 16:36:48 2023 +0200 Remove serde cruft commit 582e7d566a Author: durch <durch@users.noreply.github.com> Date: Wed Apr 19 16:24:09 2023 +0200 Outfox framing commit 6464da5f01 Author: durch <durch@users.noreply.github.com> Date: Tue Apr 18 22:23:02 2023 +0200 Framing compat commit d5e77e499b Author: durch <durch@users.noreply.github.com> Date: Tue Apr 18 18:18:54 2023 +0200 Framed encoding serde POC commit f086f9c35a Author: durch <durch@users.noreply.github.com> Date: Tue Apr 18 16:54:21 2023 +0200 Experiment with serde * Client tweaks * Speed up from_plaintext * SurbAcks * More work on the reciever end, and outfox format * Cleanup and fmt * Wrap up rebase * Happy clippy * Fix lock files * Final cleanup
2023-05-29 10:05:11 +02:00
parent 5bd87bdaa8
commit b5c8b69547
89 changed files with 7566 additions and 955 deletions
@@ -9,8 +9,8 @@ use crate::node::TaskClient;
 use futures::StreamExt;
 use nym_mixnode_common::measure;
 use nym_sphinx::forwarding::packet::MixPacket;
-use nym_sphinx::framing::codec::SphinxCodec;
-use nym_sphinx::framing::packet::FramedSphinxPacket;
+use nym_sphinx::framing::codec::NymCodec;
+use nym_sphinx::framing::packet::FramedNymPacket;
 use nym_sphinx::Delay as SphinxDelay;
 use std::net::SocketAddr;
 use tokio::net::TcpStream;
@@ -54,7 +54,7 @@ impl ConnectionHandler {
        feature = "cpucycles",
        instrument(skip(self, framed_sphinx_packet), fields(cpucycles))
    )]
-    fn handle_received_packet(&self, framed_sphinx_packet: FramedSphinxPacket) {
+    fn handle_received_packet(&self, framed_sphinx_packet: FramedNymPacket) {
        //
        // TODO: here be replay attack detection - it will require similar key cache to the one in
        // packet processor for vpn packets,
@@ -86,7 +86,7 @@ impl ConnectionHandler {
    ) {
        debug!("Starting connection handler for {:?}", remote);
        shutdown.mark_as_success();
-        let mut framed_conn = Framed::new(conn, SphinxCodec);
+        let mut framed_conn = Framed::new(conn, NymCodec);
        while !shutdown.is_shutdown() {
            tokio::select! {
                biased;
@@ -108,7 +108,7 @@ impl ConnectionHandler {
                        }
                        Some(Err(err)) => {
                            error!(
-                                "The socket connection got corrupted with error: {err}. Closing the socket",
+                                "{remote:?} - The socket connection got corrupted with error: {err}. Closing the socket",
                            );
                            return;
                        }
@@ -6,7 +6,7 @@ use nym_crypto::asymmetric::encryption;
 use nym_mixnode_common::packet_processor::error::MixProcessingError;
 pub use nym_mixnode_common::packet_processor::processor::MixProcessingResult;
 use nym_mixnode_common::packet_processor::processor::SphinxPacketProcessor;
-use nym_sphinx::framing::packet::FramedSphinxPacket;
+use nym_sphinx::framing::packet::FramedNymPacket;

 // PacketProcessor contains all data required to correctly unwrap and forward sphinx packets
 #[derive(Clone)]
@@ -31,7 +31,7 @@ impl PacketProcessor {

    pub(crate) fn process_received(
        &self,
-        received: FramedSphinxPacket,
+        received: FramedNymPacket,
    ) -> Result<MixProcessingResult, MixProcessingError> {
        self.node_stats_update_sender.report_received();
        self.inner_processor.process_received(received)
@@ -58,12 +58,12 @@ where

    fn forward_packet(&mut self, packet: MixPacket) {
        let next_hop = packet.next_hop();
-        let packet_mode = packet.packet_mode();
-        let sphinx_packet = packet.into_sphinx_packet();
+        let packet_type = packet.packet_type();
+        let packet = packet.into_packet();

-        if let Err(err) =
-            self.mixnet_client
-                .send_without_response(next_hop, sphinx_packet, packet_mode)
+        if let Err(err) = self
+            .mixnet_client
+            .send_without_response(next_hop, packet, packet_type)
        {
            if err.kind() == io::ErrorKind::WouldBlock {
                // we only know for sure if we dropped a packet if our sending queue was full
@@ -134,38 +134,38 @@ mod tests {
    use std::sync::{Arc, Mutex};
    use std::time::Duration;

+    use nym_sphinx::NymPacket;
    use nym_task::TaskManager;

    use nym_sphinx::addressing::nodes::NymNodeRoutingAddress;
    use nym_sphinx_params::packet_sizes::PacketSize;
-    use nym_sphinx_params::PacketMode;
-    use nym_sphinx_types::builder::SphinxPacketBuilder;
+    use nym_sphinx_params::PacketType;
    use nym_sphinx_types::{
        crypto, Delay as SphinxDelay, Destination, DestinationAddressBytes, Node, NodeAddressBytes,
-        SphinxPacket, DESTINATION_ADDRESS_LENGTH, IDENTIFIER_LENGTH, NODE_ADDRESS_LENGTH,
+        DESTINATION_ADDRESS_LENGTH, IDENTIFIER_LENGTH, NODE_ADDRESS_LENGTH,
    };

    #[derive(Default)]
    struct TestClient {
-        pub packets_sent: Arc<Mutex<Vec<(NymNodeRoutingAddress, SphinxPacket, PacketMode)>>>,
+        pub packets_sent: Arc<Mutex<Vec<(NymNodeRoutingAddress, NymPacket, PacketType)>>>,
    }

    impl nym_mixnet_client::SendWithoutResponse for TestClient {
        fn send_without_response(
            &mut self,
            address: NymNodeRoutingAddress,
-            packet: SphinxPacket,
-            packet_mode: PacketMode,
+            packet: NymPacket,
+            packet_type: PacketType,
        ) -> io::Result<()> {
            self.packets_sent
                .lock()
                .unwrap()
-                .push((address, packet, packet_mode));
+                .push((address, packet, packet_type));
            Ok(())
        }
    }

-    fn make_valid_sphinx_packet(size: PacketSize) -> SphinxPacket {
+    fn make_valid_sphinx_packet(size: PacketSize) -> NymPacket {
        let (_, node1_pk) = crypto::keygen();
        let node1 = Node::new(
            NodeAddressBytes::from_bytes([5u8; NODE_ADDRESS_LENGTH]),
@@ -192,12 +192,45 @@ mod tests {
            SphinxDelay::new_from_nanos(42),
            SphinxDelay::new_from_nanos(42),
        ];
-        SphinxPacketBuilder::new()
-            .with_payload_size(size.payload_size())
-            .build_packet(b"foomp", &route, &destination, &delays)
+        NymPacket::sphinx_build(size.payload_size(), b"foomp", &route, &destination, &delays)
            .unwrap()
    }

+    fn make_valid_outfox_packet(size: PacketSize) -> NymPacket {
+        let (_, node1_pk) = crypto::keygen();
+        let node1 = Node::new(
+            NodeAddressBytes::from_bytes([5u8; NODE_ADDRESS_LENGTH]),
+            node1_pk,
+        );
+        let (_, node2_pk) = crypto::keygen();
+        let node2 = Node::new(
+            NodeAddressBytes::from_bytes([4u8; NODE_ADDRESS_LENGTH]),
+            node2_pk,
+        );
+        let (_, node3_pk) = crypto::keygen();
+        let node3 = Node::new(
+            NodeAddressBytes::from_bytes([2u8; NODE_ADDRESS_LENGTH]),
+            node3_pk,
+        );
+
+        let (_, node4_pk) = crypto::keygen();
+        let node4 = Node::new(
+            NodeAddressBytes::from_bytes([2u8; NODE_ADDRESS_LENGTH]),
+            node4_pk,
+        );
+
+        let destination = Destination::new(
+            DestinationAddressBytes::from_bytes([3u8; DESTINATION_ADDRESS_LENGTH]),
+            [4u8; IDENTIFIER_LENGTH],
+        );
+
+        let route = &[node1, node2, node3, node4];
+
+        let payload = vec![1; 48];
+
+        NymPacket::outfox_build(payload, route, &destination, Some(size.plaintext_size())).unwrap()
+    }
+
    #[tokio::test]
    async fn packets_received_are_forwarded() {
        // Wire up the DelayForwarder
@@ -219,7 +252,50 @@ mod tests {
        let mix_packet = MixPacket::new(
            next_hop,
            make_valid_sphinx_packet(PacketSize::default()),
-            PacketMode::default(),
+            PacketType::default(),
+        );
+        let forward_instant = None;
+        packet_sender
+            .unbounded_send((mix_packet, forward_instant))
+            .unwrap();
+
+        // Give the the worker a chance to act
+        tokio::time::sleep(Duration::from_millis(10)).await;
+
+        // The client should have forwarded the packet straight away
+        assert_eq!(
+            client_packets_sent
+                .lock()
+                .unwrap()
+                .iter()
+                .map(|(a, _, _)| *a)
+                .collect::<Vec<_>>(),
+            vec![next_hop]
+        );
+    }
+
+    #[tokio::test]
+    async fn outfox_packets_received_are_forwarded() {
+        // Wire up the DelayForwarder
+        let (stats_sender, _stats_receiver) = mpsc::unbounded();
+        let node_stats_update_sender = UpdateSender::new(stats_sender);
+        let client = TestClient::default();
+        let client_packets_sent = client.packets_sent.clone();
+        let shutdown = TaskManager::default();
+        let mut delay_forwarder =
+            DelayForwarder::new(client, node_stats_update_sender, shutdown.subscribe());
+        let packet_sender = delay_forwarder.sender();
+
+        // Spawn the worker, listening on packet_sender channel
+        tokio::spawn(async move { delay_forwarder.run().await });
+
+        // Send a `MixPacket` down the channel without any delay attached.
+        let next_hop =
+            NymNodeRoutingAddress::from(SocketAddr::new(IpAddr::V4(Ipv4Addr::new(1, 2, 3, 4)), 42));
+        let mix_packet = MixPacket::new(
+            next_hop,
+            make_valid_outfox_packet(PacketSize::default()),
+            PacketType::default(),
        );
        let forward_instant = None;
        packet_sender