diff --git a/Cargo.lock b/Cargo.lock index 188924826d..f46bb76fe5 100644 --- a/Cargo.lock +++ b/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", diff --git a/common/client-core/Cargo.toml b/common/client-core/Cargo.toml index 773a7402ba..a3c877be83 100644 --- a/common/client-core/Cargo.toml +++ b/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 } diff --git a/common/client-core/src/client/inbound_messages.rs b/common/client-core/src/client/inbound_messages.rs index 9ddc981d64..c6fc5b03b5 100644 --- a/common/client-core/src/client/inbound_messages.rs +++ b/common/client-core/src/client/inbound_messages.rs @@ -1,16 +1,20 @@ +use hyper::body::Buf; // Copyright 2020-2023 - Nym Technologies SA // 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; pub type InputMessageReceiver = tokio::sync::mpsc::Receiver; -#[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 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, 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)) + } +} diff --git a/common/nymsphinx/anonymous-replies/src/requests.rs b/common/nymsphinx/anonymous-replies/src/requests.rs index 9dd4c84dc0..1069543926 100644 --- a/common/nymsphinx/anonymous-replies/src/requests.rs +++ b/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]); diff --git a/common/nymsphinx/forwarding/Cargo.toml b/common/nymsphinx/forwarding/Cargo.toml index c3f3e0dc9f..efa5b2e1fb 100644 --- a/common/nymsphinx/forwarding/Cargo.toml +++ b/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 } \ No newline at end of file diff --git a/common/nymsphinx/forwarding/src/packet.rs b/common/nymsphinx/forwarding/src/packet.rs index 97b1419aa1..7846e1e067 100644 --- a/common/nymsphinx/forwarding/src/packet.rs +++ b/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, 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(&self, serializer: S) -> Result { + 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(self, v: &[u8]) -> Result { + MixPacket::try_from_bytes(v).map_err(serde::de::Error::custom) + } +} + +impl <'de> Deserialize <'de> for MixPacket { + fn deserialize>(deserializer: D) -> Result { + deserializer.deserialize_bytes(MixPacketVisitor) + } } // TODO: test for serialization and errors! diff --git a/common/task/Cargo.toml b/common/task/Cargo.toml index 6f0f6b7579..65095d6797 100644 --- a/common/task/Cargo.toml +++ b/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 diff --git a/common/task/src/connections.rs b/common/task/src/connections.rs index af13111526..e55175f1eb 100644 --- a/common/task/src/connections.rs +++ b/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 diff --git a/sdk/rust/nym-sdk/Cargo.toml b/sdk/rust/nym-sdk/Cargo.toml index b657db8570..98318604ae 100644 --- a/sdk/rust/nym-sdk/Cargo.toml +++ b/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" } diff --git a/sdk/rust/nym-sdk/src/mixnet/native_client.rs b/sdk/rust/nym-sdk/src/mixnet/native_client.rs index 2f9777cd2f..5620d58980 100644 --- a/sdk/rust/nym-sdk/src/mixnet/native_client.rs +++ b/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 for MixnetClient { + type Error = Error; + + fn poll_ready(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + 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> { + 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> { + self.client_input.input_sender.poll_close_unpin(cx).map_err(|_| Error::MessageSendingFailure) + } +} + impl Stream for MixnetClient { type Item = ReconstructedMessage;