InputMessageCodec, Serde for MixPacket

Cargo.lock

@@ -4138,6 +4138,7 @@ version = "1.1.15"
 dependencies = [
  "async-trait",
  "base64 0.21.7",
+ "bincode",
  "bs58 0.5.1",
  "cfg-if",
  "clap 4.5.4",
@@ -5466,6 +5467,7 @@ dependencies = [
  "nym-sphinx-addressing",
  "nym-sphinx-params",
  "nym-sphinx-types",
+ "serde",
  "thiserror",
 ]
 
@@ -5543,6 +5545,7 @@ version = "0.1.0"
 dependencies = [
  "futures",
  "log",
+ "serde",
  "thiserror",
  "tokio",
  "wasm-bindgen",

common/client-core/Cargo.toml

@@ -10,6 +10,7 @@ license.workspace = true
 
 [dependencies]
 async-trait = { workspace = true }
+bincode = { workspace = true }
 base64 = "0.21.2"
 bs58 = { workspace = true }
 cfg-if = { workspace = true }
@@ -26,7 +27,7 @@ tap = { workspace = true }
 thiserror = { workspace = true }
 url = { workspace = true, features = ["serde"] }
 tokio = { workspace = true, features = ["macros"] }
-tokio-util = { workspace = true }
+tokio-util = { workspace = true, features = ["codec"] }
 time = { workspace = true }
 zeroize = { workspace = true }
 

common/client-core/src/client/inbound_messages.rs

@@ -1,16 +1,20 @@
+use hyper::body::Buf;
 // Copyright 2020-2023 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
-
 use nym_sphinx::addressing::clients::Recipient;
 use nym_sphinx::anonymous_replies::requests::AnonymousSenderTag;
 use nym_sphinx::forwarding::packet::MixPacket;
 use nym_sphinx::params::PacketType;
 use nym_task::connections::TransmissionLane;
+use serde::{Deserialize, Serialize};
+use tokio_util::{bytes::BytesMut, codec::{Decoder, Encoder}};
+
+use crate::error::ClientCoreError;
 
 pub type InputMessageSender = tokio_util::sync::PollSender<InputMessage>;
 pub type InputMessageReceiver = tokio::sync::mpsc::Receiver<InputMessage>;
 
-#[derive(Debug)]
+#[derive(Serialize, Deserialize, Debug)]
 pub enum InputMessage {
     /// Fire an already prepared mix packets into the network.
     /// No guarantees are made about it. For example no retransmssion
@@ -198,3 +202,46 @@ impl InputMessage {
         }
     }
 }
+
+
+// TODO: Tests
+pub struct InputMessageCodec;
+
+impl Encoder<InputMessage> for InputMessageCodec {
+    type Error = ClientCoreError;
+
+    fn encode(&mut self, item: InputMessage, buf: &mut BytesMut) -> Result<(), Self::Error> {
+        let encoded = bincode::serialize(&item).expect("failed to serialize InputMessage");
+        let encoded_len = encoded.len() as u32;
+        let mut encoded_with_len = encoded_len.to_le_bytes().to_vec();
+        encoded_with_len.extend(encoded);
+        buf.reserve(encoded_with_len.len());
+        buf.extend_from_slice(&encoded_with_len);
+        Ok(())
+    }
+}
+
+impl Decoder for InputMessageCodec {
+    type Item = InputMessage;
+    type Error = ClientCoreError;
+
+    fn decode(&mut self, buf: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
+        if buf.len() < 4 {
+            return Ok(None);
+        }
+
+        let len = u32::from_le_bytes([buf[0], buf[1], buf[2], buf[3]]) as usize;
+        if buf.len() < len + 4 {
+            return Ok(None);
+        }
+
+        let decoded = match bincode::deserialize(&buf[4..len]) {
+            Ok(decoded) => decoded,
+            Err(_) => return Ok(None)
+        };
+
+        buf.advance(len + 4);
+
+        Ok(Some(decoded))
+    }
+}

common/nymsphinx/anonymous-replies/src/requests.rs

@@ -4,6 +4,7 @@
 use crate::{ReplySurb, ReplySurbError};
 use nym_sphinx_addressing::clients::{Recipient, RecipientFormattingError};
 use rand::{CryptoRng, RngCore};
+use serde::{Deserialize, Serialize};
 use std::fmt::{Display, Formatter};
 use std::mem;
 use thiserror::Error;
@@ -24,7 +25,7 @@ pub enum InvalidAnonymousSenderTagRepresentation {
     InvalidLength { received: usize, expected: usize },
 }
 
-#[derive(Debug, Copy, Clone, Eq, PartialEq, Hash)]
+#[derive(Debug, Copy, Clone, Eq, PartialEq, Hash, Serialize, Deserialize)]
 #[cfg_attr(target_arch = "wasm32", wasm_bindgen)]
 pub struct AnonymousSenderTag([u8; SENDER_TAG_SIZE]);
 

common/nymsphinx/forwarding/Cargo.toml

@@ -13,3 +13,4 @@ nym-sphinx-params = { path = "../params" }
 nym-sphinx-types = { path = "../types", features = ["sphinx", "outfox"] }
 nym-outfox = { path = "../../../nym-outfox" }
 thiserror = { workspace = true }
+serde = { workspace = true }

common/nymsphinx/forwarding/src/packet.rs

@@ -4,6 +4,7 @@
 use nym_sphinx_addressing::nodes::{NymNodeRoutingAddress, NymNodeRoutingAddressError};
 use nym_sphinx_params::{PacketSize, PacketType};
 use nym_sphinx_types::{NymPacket, NymPacketError};
+use serde::{de::{self, Visitor}, Deserialize, Deserializer, Serialize, Serializer};
 
 use std::fmt::{self, Debug, Formatter};
 use thiserror::Error;
@@ -110,6 +111,39 @@ impl MixPacket {
             .chain(self.packet.to_bytes()?)
             .collect())
     }
+
+    pub fn to_bytes(&self) -> Result<Vec<u8>, MixPacketFormattingError> {
+        Ok(std::iter::once(self.packet_type as u8)
+            .chain(self.next_hop.as_bytes())
+            .chain(self.packet.to_bytes()?)
+            .collect())
+    }
+}
+
+impl Serialize for MixPacket {
+    fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
+        serializer.serialize_bytes(&self.to_bytes().map_err(serde::ser::Error::custom)?)
+    }
+}
+
+struct MixPacketVisitor;
+
+impl <'de> Visitor<'de> for MixPacketVisitor {
+    type Value = MixPacket;
+
+    fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
+        formatter.write_str("a byte array representing a mix packet")
+    }
+
+    fn visit_bytes<E: de::Error>(self, v: &[u8]) -> Result<Self::Value, E> {
+        MixPacket::try_from_bytes(v).map_err(serde::de::Error::custom)
+    }
+}
+
+impl <'de> Deserialize <'de> for MixPacket {
+    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
+        deserializer.deserialize_bytes(MixPacketVisitor)
+    }
 }
 
 // TODO: test for serialization and errors!

common/task/Cargo.toml

@@ -12,6 +12,7 @@ futures = { workspace = true }
 log = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true, features = ["macros", "sync"] }
+serde = { workspace = true }
 
 [target."cfg(not(target_arch = \"wasm32\"))".dependencies.tokio]
 workspace = true

common/task/src/connections.rs

@@ -2,11 +2,12 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use futures::channel::mpsc;
+use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 
 pub type ConnectionId = u64;
 
-#[derive(Copy, Clone, Debug, Hash, PartialEq, Eq)]
+#[derive(Copy, Clone, Debug, Hash, PartialEq, Eq, Serialize, Deserialize)]
 pub enum TransmissionLane {
     General,
     // we need to treat surb-related requests and responses at higher priority

sdk/rust/nym-sdk/Cargo.toml

@@ -9,7 +9,10 @@ license.workspace = true
 [dependencies]
 async-trait = { workspace = true }
 bip39 = { workspace = true }
-nym-client-core = { path = "../../../common/client-core", features = ["fs-surb-storage", "fs-gateways-storage"] }
+nym-client-core = { path = "../../../common/client-core", features = [
+    "fs-surb-storage",
+    "fs-gateways-storage",
+] }
 nym-crypto = { path = "../../../common/crypto" }
 nym-gateway-requests = { path = "../../../gateway/gateway-requests" }
 nym-bandwidth-controller = { path = "../../../common/bandwidth-controller" }
@@ -21,7 +24,9 @@ nym-sphinx = { path = "../../../common/nymsphinx" }
 nym-task = { path = "../../../common/task" }
 nym-topology = { path = "../../../common/topology" }
 nym-socks5-client-core = { path = "../../../common/socks5-client-core" }
-nym-validator-client = { path = "../../../common/client-libs/validator-client", features = ["http-client"] }
+nym-validator-client = { path = "../../../common/client-libs/validator-client", features = [
+    "http-client",
+] }
 nym-socks5-requests = { path = "../../../common/socks5/requests" }
 nym-ordered-buffer = { path = "../../../common/socks5/ordered-buffer" }
 nym-service-providers-common = { path = "../../../service-providers/common" }

sdk/rust/nym-sdk/src/mixnet/native_client.rs

@@ -2,7 +2,7 @@ use crate::mixnet::client::MixnetClientBuilder;
 use crate::mixnet::traits::MixnetMessageSender;
 use crate::{Error, Result};
 use async_trait::async_trait;
-use futures::{ready, AsyncRead, Stream, StreamExt};
+use futures::{ready, AsyncRead, Sink, SinkExt, Stream, StreamExt};
 use log::error;
 use nym_client_core::client::base_client::GatewayConnection;
 use nym_client_core::client::{
@@ -258,6 +258,30 @@ impl AsyncRead for MixnetClient {
     }
 }
 
+impl Sink<InputMessage> for MixnetClient {
+    type Error = Error;
+
+    fn poll_ready(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>> {
+        match self.client_input.input_sender.poll_ready_unpin(cx) {
+            Poll::Ready(Ok(())) => Poll::Ready(Ok(())),
+            Poll::Ready(Err(_)) => Poll::Ready(Err(Error::MessageSendingFailure)),
+            Poll::Pending => Poll::Pending,
+        }
+    }
+
+    fn start_send(mut self: Pin<&mut Self>, item: InputMessage) -> Result<()> {
+        self.client_input.input_sender.start_send_unpin(item).map_err(|_| Error::MessageSendingFailure)
+    }
+
+    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>> {
+        self.client_input.input_sender.poll_flush_unpin(cx).map_err(|_| Error::MessageSendingFailure)
+    }
+
+    fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>> {
+        self.client_input.input_sender.poll_close_unpin(cx).map_err(|_| Error::MessageSendingFailure)
+    }
+}
+
 impl Stream for MixnetClient {
     type Item = ReconstructedMessage;