diff --git a/common/client-core/src/client/inbound_messages.rs b/common/client-core/src/client/inbound_messages.rs index 4fd7ad134b..66e0b49a86 100644 --- a/common/client-core/src/client/inbound_messages.rs +++ b/common/client-core/src/client/inbound_messages.rs @@ -6,6 +6,7 @@ use nym_sphinx::forwarding::packet::MixPacket; use nym_sphinx::params::PacketType; use nym_task::connections::TransmissionLane; use serde::{Deserialize, Serialize}; +use std::convert::TryInto; use tokio_util::{ bytes::Buf, bytes::BytesMut, @@ -17,6 +18,8 @@ use crate::error::ClientCoreError; pub type InputMessageSender = tokio_util::sync::PollSender; pub type InputMessageReceiver = tokio::sync::mpsc::Receiver; +const LENGHT_ENCODING_PREFIX_SIZE: usize = 4; + #[derive(Serialize, Deserialize, Debug)] pub enum InputMessage { /// Fire an already prepared mix packets into the network. @@ -198,11 +201,25 @@ impl InputMessage { } pub fn serialized_size(&self) -> u64 { - bincode::serialized_size(self).expect("failed to get serialized InputMessage size") + 4 + bincode::serialized_size(self).expect("failed to get serialized InputMessage size") + + LENGHT_ENCODING_PREFIX_SIZE as u64 } } // TODO: Tests +pub struct AdressedInputMessageCodec(pub Recipient); + +impl Encoder<&[u8]> for AdressedInputMessageCodec { + type Error = ClientCoreError; + + fn encode(&mut self, item: &[u8], buf: &mut BytesMut) -> Result<(), Self::Error> { + let mut codec = InputMessageCodec; + let input_message = InputMessage::simple(item, self.0.clone()); + codec.encode(input_message, buf)?; + Ok(()) + } +} + pub struct InputMessageCodec; impl Encoder for InputMessageCodec { @@ -224,21 +241,27 @@ impl Decoder for InputMessageCodec { type Error = ClientCoreError; fn decode(&mut self, buf: &mut BytesMut) -> Result, Self::Error> { - if buf.len() < 4 { + if buf.len() < LENGHT_ENCODING_PREFIX_SIZE { return Ok(None); } - let len = u32::from_le_bytes([buf[0], buf[1], buf[2], buf[3]]) as usize; - if buf.len() < len + 4 { + let len = u32::from_le_bytes( + buf[0..LENGHT_ENCODING_PREFIX_SIZE] + .try_into() + .expect("Could not coarce to array"), + ) as usize; + if buf.len() < len + LENGHT_ENCODING_PREFIX_SIZE { return Ok(None); } - let decoded = match bincode::deserialize(&buf[4..len + 4]) { + let decoded = match bincode::deserialize( + &buf[LENGHT_ENCODING_PREFIX_SIZE..len + LENGHT_ENCODING_PREFIX_SIZE], + ) { Ok(decoded) => decoded, Err(_) => return Ok(None), }; - buf.advance(len + 4); + buf.advance(len + LENGHT_ENCODING_PREFIX_SIZE); Ok(Some(decoded)) } diff --git a/common/nymsphinx/src/receiver.rs b/common/nymsphinx/src/receiver.rs index c637c16e73..e10f8803a4 100644 --- a/common/nymsphinx/src/receiver.rs +++ b/common/nymsphinx/src/receiver.rs @@ -63,7 +63,7 @@ impl From for ReconstructedMessage { } pub struct ReconstructedMessageCodec; -const OFFSET: usize = 4; +const LENGHT_ENCODING_PREFIX_SIZE: usize = 4; impl Encoder for ReconstructedMessageCodec { type Error = MessageRecoveryError; @@ -88,21 +88,23 @@ impl Decoder for ReconstructedMessageCodec { type Error = MessageRecoveryError; fn decode(&mut self, buf: &mut BytesMut) -> Result, Self::Error> { - if buf.len() < OFFSET { + if buf.len() < LENGHT_ENCODING_PREFIX_SIZE { return Ok(None); } let len = u32::from_le_bytes( - buf[0..OFFSET] + buf[0..LENGHT_ENCODING_PREFIX_SIZE] .try_into() - .expect("We know that we have at least OFFSET bytes in there"), + .expect("We know that we have at least LENGHT_ENCODING_PREFIX_SIZE bytes in there"), ) as usize; - if buf.len() < len + OFFSET { + if buf.len() < len + LENGHT_ENCODING_PREFIX_SIZE { return Ok(None); } - let decoded = match bincode::deserialize(&buf[OFFSET..len + OFFSET]) { + let decoded = match bincode::deserialize( + &buf[LENGHT_ENCODING_PREFIX_SIZE..len + LENGHT_ENCODING_PREFIX_SIZE], + ) { Ok(decoded) => decoded, Err(e) => { debug!("Failed to decode the message - {:?}", e); @@ -110,7 +112,7 @@ impl Decoder for ReconstructedMessageCodec { } }; - buf.advance(len + 4); + buf.advance(len + LENGHT_ENCODING_PREFIX_SIZE); Ok(Some(decoded)) } diff --git a/sdk/rust/nym-sdk/src/mixnet/native_client.rs b/sdk/rust/nym-sdk/src/mixnet/native_client.rs index 4b1bfcd9c9..bc7df5ea49 100644 --- a/sdk/rust/nym-sdk/src/mixnet/native_client.rs +++ b/sdk/rust/nym-sdk/src/mixnet/native_client.rs @@ -384,7 +384,7 @@ impl MixnetClient { Poll::Ready(Ok(())) } else { let written = buf.capacity(); - buf.put_slice(&self._read.buffer[..written]); + buf.put_slice(&self._read.buffer.split_off(written)); self._read.buffer.advance(written); cx.waker().wake_by_ref(); Poll::Ready(Ok(())) @@ -426,11 +426,16 @@ impl AsyncRead for MixnetClient { Poll::Pending => return Poll::Pending, }; - // let mut buffer = BytesMut::new(); - - codec.encode(msg, &mut self._read.buffer).unwrap(); - - // = buffer.to_vec(); + match codec.encode(msg, &mut self._read.buffer) { + Ok(_) => {} + Err(e) => { + error!("failed to encode reconstructed message: {:?}", e); + return Poll::Ready(Err(tokio::io::Error::new( + tokio::io::ErrorKind::Other, + "failed to encode reconstructed message", + ))); + } + }; self.read_buffer_to_slice(buf, cx) }