diff --git a/Cargo.lock b/Cargo.lock index 47aa3895a1..86c58f5283 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3949,7 +3949,6 @@ dependencies = [ "nym-credential-storage", "nym-crypto", "nym-network-defaults", - "nym-ordered-buffer", "nym-sdk", "nym-service-providers-common", "nym-socks5-proxy-helpers", diff --git a/common/socks5-client-core/src/socks/client.rs b/common/socks5-client-core/src/socks/client.rs index ebd980a0e6..e630a4b392 100644 --- a/common/socks5-client-core/src/socks/client.rs +++ b/common/socks5-client-core/src/socks/client.rs @@ -428,18 +428,14 @@ impl SocksClient { Some(self.lane_queue_lengths.clone()), self.shutdown_listener.clone(), ) - .run(move |conn_id, read_data, socket_closed| { - let provider_request = Socks5Request::new_send( - request_version.provider_protocol, - conn_id, - read_data, - socket_closed, - ); + .run(move |socket_data| { + let lane = TransmissionLane::ConnectionId(socket_data.header.connection_id); + let provider_request = + Socks5Request::new_send(request_version.provider_protocol, socket_data); let provider_message = Socks5ProviderRequest::new_provider_data( request_version.provider_interface, provider_request, ); - let lane = TransmissionLane::ConnectionId(conn_id); if anonymous { InputMessage::new_anonymous( recipient, diff --git a/common/socks5-client-core/src/socks/mixnet_responses.rs b/common/socks5-client-core/src/socks/mixnet_responses.rs index d8ef425ddb..05cb5bd48d 100644 --- a/common/socks5-client-core/src/socks/mixnet_responses.rs +++ b/common/socks5-client-core/src/socks/mixnet_responses.rs @@ -1,19 +1,20 @@ +// Copyright 2020-2023 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::error::Socks5ClientCoreError; use futures::channel::mpsc; use futures::StreamExt; use log::*; - use nym_client_core::client::received_buffer::ReconstructedMessagesReceiver; use nym_client_core::client::received_buffer::{ ReceivedBufferMessage, ReceivedBufferRequestSender, }; use nym_service_providers_common::interface::{ControlResponse, ResponseContent}; -use nym_socks5_proxy_helpers::connection_controller::ControllerSender; +use nym_socks5_proxy_helpers::connection_controller::{ControllerCommand, ControllerSender}; use nym_socks5_requests::{Socks5ProviderResponse, Socks5Response, Socks5ResponseContent}; use nym_sphinx::receiver::ReconstructedMessage; use nym_task::TaskClient; -use crate::error::Socks5ClientCoreError; - pub(crate) struct MixnetResponseListener { buffer_requester: ReceivedBufferRequestSender, mix_response_receiver: ReconstructedMessagesReceiver, @@ -79,9 +80,9 @@ impl MixnetResponseListener { ); Err(err_response.into()) } - Socks5ResponseContent::NetworkData(response) => { + Socks5ResponseContent::NetworkData { content } => { self.controller_sender - .unbounded_send(response.into()) + .unbounded_send(ControllerCommand::new_send(content)) .unwrap(); Ok(()) } diff --git a/common/socks5/ordered-buffer/src/buffer.rs b/common/socks5/ordered-buffer/src/buffer.rs index 33c4d69142..1b7b281727 100644 --- a/common/socks5/ordered-buffer/src/buffer.rs +++ b/common/socks5/ordered-buffer/src/buffer.rs @@ -1,6 +1,21 @@ -use crate::message::OrderedMessage; +// Copyright 2020-2023 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + use log::*; -use std::collections::HashMap; +use std::collections::BTreeMap; +use thiserror::Error; + +#[derive(Debug, Error, PartialEq, Eq)] +pub enum OrderedMessageError { + #[error("received message with sequence number {received}, which is way higher than our current {current}")] + MessageSequenceTooLarge { current: u64, received: u64 }, + + #[error("received message with sequence number {received}, while we're already at {current}!")] + MessageAlreadyReconstructed { current: u64, received: u64 }, + + #[error("attempted to overwrite message at sequence {received}")] + AttemptedToOverwriteSequence { received: u64 }, +} /// Stores messages and emits them in order. /// @@ -9,36 +24,58 @@ use std::collections::HashMap; /// to fill up with the full sequence. #[derive(Debug)] pub struct OrderedMessageBuffer { - next_index: u64, - messages: HashMap, + next_sequence: u64, + messages: BTreeMap>, } /// Data returned from `OrderedMessageBuffer` on a successful read of gapless ordered data. #[derive(Debug, PartialEq, Eq)] pub struct ReadContiguousData { pub data: Vec, - pub last_index: u64, + pub last_sequence: u64, } +const MAX_REASONABLE_OFFSET: u64 = 1000; + impl OrderedMessageBuffer { pub fn new() -> OrderedMessageBuffer { OrderedMessageBuffer { - next_index: 0, - messages: HashMap::new(), + next_sequence: 0, + messages: BTreeMap::new(), } } /// Writes a message to the buffer. messages are sort on insertion, so /// that later on multiple reads for incomplete sequences don't result in /// useless sort work. - pub fn write(&mut self, message: OrderedMessage) { + pub fn write(&mut self, sequence: u64, data: Vec) -> Result<(), OrderedMessageError> { + // reject messages that have clearly malformed sequence + if sequence > self.next_sequence + MAX_REASONABLE_OFFSET { + return Err(OrderedMessageError::MessageSequenceTooLarge { + current: self.next_sequence, + received: sequence, + }); + } + + if self.messages.contains_key(&sequence) { + return Err(OrderedMessageError::AttemptedToOverwriteSequence { received: sequence }); + } + + if sequence < self.next_sequence { + return Err(OrderedMessageError::MessageAlreadyReconstructed { + current: self.next_sequence, + received: sequence, + }); + } + trace!( - "Writing message index: {} length {:?} to OrderedMessageBuffer.", - message.index, - message.data.len() + "Writing message index: {} length {} to OrderedMessageBuffer.", + sequence, + data.len() ); - self.messages.insert(message.index, message); + self.messages.insert(sequence, data); + Ok(()) } /// Returns `Option>` where it's `Some(bytes)` if there is gapless @@ -49,33 +86,31 @@ impl OrderedMessageBuffer { /// a read will return the bytes of messages 0, 1, 2. Subsequent reads will /// return `None` until message 3 comes in, at which point 3, 4, and any /// further contiguous messages which have arrived will be returned. + #[must_use] pub fn read(&mut self) -> Option { - if !self.messages.contains_key(&self.next_index) { + if !self.messages.contains_key(&self.next_sequence) { return None; } let mut contiguous_messages = Vec::new(); - let mut index = self.next_index; + let mut seq = self.next_sequence; - while let Some(ordered_message) = self.messages.remove(&index) { - contiguous_messages.push(ordered_message); - index += 1; + while let Some(mut data) = self.messages.remove(&seq) { + contiguous_messages.append(&mut data); + seq += 1; } - let high_water = index; - self.next_index = high_water; - trace!("Next high water mark is: {}", high_water); + let high_water = seq; + self.next_sequence = high_water; + trace!("Next high water mark is: {high_water}"); - // dig out the bytes from inside the struct - let data: Vec = contiguous_messages - .into_iter() - .flat_map(|message| message.data) - .collect(); - - trace!("Returning {} bytes from ordered message buffer", data.len()); + trace!( + "Returning {} bytes from ordered message buffer", + contiguous_messages.len() + ); Some(ReadContiguousData { - data, - last_index: index, + data: contiguous_messages, + last_sequence: seq, }) } } @@ -90,6 +125,64 @@ impl Default for OrderedMessageBuffer { mod test_chunking_and_reassembling { use super::*; + #[test] + fn trying_to_write_unreasonable_high_sequence() { + let mut buffer = OrderedMessageBuffer::new(); + let first_message = vec![1, 2, 3, 4]; + let second_message = vec![5, 6, 7, 8]; + + buffer.write(0, first_message).unwrap(); + buffer.write(1, second_message).unwrap(); + + assert_eq!( + Err(OrderedMessageError::MessageSequenceTooLarge { + current: 0, + received: 12345678 + }), + buffer.write(12345678, b"foomp".to_vec()) + ) + } + + #[test] + fn trying_to_overwrite_sequence() { + let mut buffer = OrderedMessageBuffer::new(); + let message = vec![1, 2, 3, 4]; + + buffer.write(0, message.clone()).unwrap(); + buffer.write(1, message.clone()).unwrap(); + buffer.write(2, message.clone()).unwrap(); + buffer.write(3, message.clone()).unwrap(); + + for seq in 0..=3 { + assert_eq!( + Err(OrderedMessageError::AttemptedToOverwriteSequence { received: seq }), + buffer.write(seq, message.clone()) + ) + } + } + + #[test] + fn writing_past_data() { + let mut buffer = OrderedMessageBuffer::new(); + let message = vec![1, 2, 3, 4]; + + buffer.write(0, message.clone()).unwrap(); + buffer.write(1, message.clone()).unwrap(); + buffer.write(2, message.clone()).unwrap(); + buffer.write(3, message.clone()).unwrap(); + let _ = buffer.read().unwrap(); + + for seq in 0..=3 { + assert_eq!( + Err(OrderedMessageError::MessageAlreadyReconstructed { + current: 4, + received: seq + }), + buffer.write(seq, message.clone()) + ) + } + } + #[cfg(test)] mod reading_from_and_writing_to_the_buffer { use super::*; @@ -102,20 +195,14 @@ mod test_chunking_and_reassembling { fn read_returns_ordered_bytes_and_resets_buffer() { let mut buffer = OrderedMessageBuffer::new(); - let first_message = OrderedMessage { - data: vec![1, 2, 3, 4], - index: 0, - }; - let second_message = OrderedMessage { - data: vec![5, 6, 7, 8], - index: 1, - }; + let first_message = vec![1, 2, 3, 4]; + let second_message = vec![5, 6, 7, 8]; - buffer.write(first_message); + buffer.write(0, first_message).unwrap(); let first_read = buffer.read().unwrap().data; assert_eq!(vec![1, 2, 3, 4], first_read); - buffer.write(second_message); + buffer.write(1, second_message).unwrap(); let second_read = buffer.read().unwrap().data; assert_eq!(vec![5, 6, 7, 8], second_read); @@ -126,17 +213,11 @@ mod test_chunking_and_reassembling { fn test_multiple_adds_stacks_up_bytes_in_the_buffer() { let mut buffer = OrderedMessageBuffer::new(); - let first_message = OrderedMessage { - data: vec![1, 2, 3, 4], - index: 0, - }; - let second_message = OrderedMessage { - data: vec![5, 6, 7, 8], - index: 1, - }; + let first_message = vec![1, 2, 3, 4]; + let second_message = vec![5, 6, 7, 8]; - buffer.write(first_message); - buffer.write(second_message); + buffer.write(0, first_message).unwrap(); + buffer.write(1, second_message).unwrap(); let second_read = buffer.read(); assert_eq!(vec![1, 2, 3, 4, 5, 6, 7, 8], second_read.unwrap().data); assert_eq!(None, buffer.read()); // second read on fully ordered result set is empty @@ -146,17 +227,11 @@ mod test_chunking_and_reassembling { fn out_of_order_adds_results_in_ordered_byte_vector() { let mut buffer = OrderedMessageBuffer::new(); - let first_message = OrderedMessage { - data: vec![1, 2, 3, 4], - index: 0, - }; - let second_message = OrderedMessage { - data: vec![5, 6, 7, 8], - index: 1, - }; + let first_message = vec![1, 2, 3, 4]; + let second_message = vec![5, 6, 7, 8]; - buffer.write(second_message); - buffer.write(first_message); + buffer.write(1, second_message).unwrap(); + buffer.write(0, first_message).unwrap(); let read = buffer.read().unwrap().data; assert_eq!(vec![1, 2, 3, 4, 5, 6, 7, 8], read); assert_eq!(None, buffer.read()); // second read on fully ordered result set is empty @@ -170,23 +245,13 @@ mod test_chunking_and_reassembling { fn setup() -> OrderedMessageBuffer { let mut buffer = OrderedMessageBuffer::new(); - let zero_message = OrderedMessage { - data: vec![0, 0, 0, 0], - index: 0, - }; - let one_message = OrderedMessage { - data: vec![1, 1, 1, 1], - index: 1, - }; + let zero_message = vec![0, 0, 0, 0]; + let one_message = vec![1, 1, 1, 1]; + let three_message = vec![3, 3, 3, 3]; - let three_message = OrderedMessage { - data: vec![3, 3, 3, 3], - index: 3, - }; - - buffer.write(zero_message); - buffer.write(one_message); - buffer.write(three_message); + buffer.write(0, zero_message).unwrap(); + buffer.write(1, one_message).unwrap(); + buffer.write(3, three_message).unwrap(); buffer } #[test] @@ -199,43 +264,31 @@ mod test_chunking_and_reassembling { assert_eq!(None, buffer.read()); // let's add another message, leaving a gap in place at index 2 - let five_message = OrderedMessage { - data: vec![5, 5, 5, 5], - index: 5, - }; - buffer.write(five_message); + let five_message = vec![5, 5, 5, 5]; + buffer.write(5, five_message).unwrap(); assert_eq!(None, buffer.read()); } #[test] fn filling_the_gap_allows_us_to_get_everything() { let mut buffer = setup(); - buffer.read(); // that burns the first two. We still have a gap before the 3s. + let _ = buffer.read(); // that burns the first two. We still have a gap before the 3s. - let two_message = OrderedMessage { - data: vec![2, 2, 2, 2], - index: 2, - }; - buffer.write(two_message); + let two_message = vec![2, 2, 2, 2]; + buffer.write(2, two_message).unwrap(); let more_ordered_bytes = buffer.read().unwrap().data; assert_eq!([2, 2, 2, 2, 3, 3, 3, 3].to_vec(), more_ordered_bytes); // let's add another message - let five_message = OrderedMessage { - data: vec![5, 5, 5, 5], - index: 5, - }; - buffer.write(five_message); + let five_message = vec![5, 5, 5, 5]; + buffer.write(5, five_message).unwrap(); assert_eq!(None, buffer.read()); // let's fill in the gap of 4s now and read again - let four_message = OrderedMessage { - data: vec![4, 4, 4, 4], - index: 4, - }; - buffer.write(four_message); + let four_message = vec![4, 4, 4, 4]; + buffer.write(4, four_message).unwrap(); assert_eq!( [4, 4, 4, 4, 5, 5, 5, 5].to_vec(), @@ -249,70 +302,47 @@ mod test_chunking_and_reassembling { #[test] fn filling_the_gap_allows_us_to_get_everything_when_last_element_is_empty() { let mut buffer = OrderedMessageBuffer::new(); - let zero_message = OrderedMessage { - data: vec![0, 0, 0, 0], - index: 0, - }; - let one_message = OrderedMessage { - data: vec![2, 2, 2, 2], - index: 1, - }; - let two_message = OrderedMessage { - data: vec![], - index: 2, - }; + let zero_message = vec![0, 0, 0, 0]; + let one_message = vec![2, 2, 2, 2]; + let two_message = vec![]; - buffer.write(zero_message); + buffer.write(0, zero_message).unwrap(); assert!(buffer.read().is_some()); // burn the buffer - buffer.write(two_message); - buffer.write(one_message); + buffer.write(2, two_message).unwrap(); + buffer.write(1, one_message).unwrap(); assert!(buffer.read().is_some()); - assert_eq!(buffer.next_index, 3); + assert_eq!(buffer.next_sequence, 3); } #[test] fn works_with_gaps_bigger_than_one() { let mut buffer = OrderedMessageBuffer::new(); - let zero_message = OrderedMessage { - data: vec![0, 0, 0, 0], - index: 0, - }; - let one_message = OrderedMessage { - data: vec![2, 2, 2, 2], - index: 1, - }; - let two_message = OrderedMessage { - data: vec![2, 2, 2, 2], - index: 2, - }; - let three_message = OrderedMessage { - data: vec![2, 2, 2, 2], - index: 3, - }; - let four_message = OrderedMessage { - data: vec![2, 2, 2, 2], - index: 4, - }; - buffer.write(zero_message); + let zero_message = vec![0, 0, 0, 0]; + let one_message = vec![2, 2, 2, 2]; + let two_message = vec![2, 2, 2, 2]; + let three_message = vec![2, 2, 2, 2]; + let four_message = vec![2, 2, 2, 2]; + + buffer.write(0, zero_message).unwrap(); assert!(buffer.read().is_some()); - assert_eq!(buffer.next_index, 1); + assert_eq!(buffer.next_sequence, 1); - buffer.write(four_message); + buffer.write(4, four_message).unwrap(); assert!(buffer.read().is_none()); - assert_eq!(buffer.next_index, 1); + assert_eq!(buffer.next_sequence, 1); - buffer.write(three_message); + buffer.write(3, three_message).unwrap(); assert!(buffer.read().is_none()); - assert_eq!(buffer.next_index, 1); + assert_eq!(buffer.next_sequence, 1); - buffer.write(two_message); + buffer.write(2, two_message).unwrap(); assert!(buffer.read().is_none()); - assert_eq!(buffer.next_index, 1); + assert_eq!(buffer.next_sequence, 1); - buffer.write(one_message); + buffer.write(1, one_message).unwrap(); assert!(buffer.read().is_some()); - assert_eq!(buffer.next_index, 5) + assert_eq!(buffer.next_sequence, 5) } } } diff --git a/common/socks5/ordered-buffer/src/lib.rs b/common/socks5/ordered-buffer/src/lib.rs index 9ed9d9c867..97e5a76c1f 100644 --- a/common/socks5/ordered-buffer/src/lib.rs +++ b/common/socks5/ordered-buffer/src/lib.rs @@ -1,8 +1,3 @@ mod buffer; -mod message; -mod sender; -pub use buffer::{OrderedMessageBuffer, ReadContiguousData}; -pub use message::MessageError; -pub use message::OrderedMessage; -pub use sender::OrderedMessageSender; +pub use buffer::{OrderedMessageBuffer, OrderedMessageError, ReadContiguousData}; diff --git a/common/socks5/ordered-buffer/src/message.rs b/common/socks5/ordered-buffer/src/message.rs deleted file mode 100644 index 15627d7fe0..0000000000 --- a/common/socks5/ordered-buffer/src/message.rs +++ /dev/null @@ -1,143 +0,0 @@ -// Copyright 2020-2022 - Nym Technologies SA -// SPDX-License-Identifier: Apache-2.0 - -use std::cmp::Ordering; -use thiserror::Error; - -#[derive(Error, Debug, PartialEq, Eq)] -pub enum MessageError { - #[error("the received message was empty")] - NoData, - - #[error("could not extract message index. Received {received} bytes, but expected {expected}")] - IndexTooShort { received: usize, expected: usize }, -} - -#[derive(Clone, Debug, Eq, PartialEq)] -pub struct OrderedMessage { - pub data: Vec, - pub index: u64, -} - -impl OrderedMessage { - /// Serializes an `OrderedMessage` into bytes. - /// The output format is: - /// | 8 bytes index | data... | - pub fn into_bytes(self) -> Vec { - self.index - .to_be_bytes() - .iter() - .cloned() - .chain(self.data.into_iter()) - .collect() - } - - /// Attempts to deserialize an `OrderedMessage` from bytes. - pub fn try_from_bytes(data: Vec) -> Result { - if data.is_empty() { - return Err(MessageError::NoData); - } - - if data.len() < 8 { - return Err(MessageError::IndexTooShort { - received: data.len(), - expected: 8, - }); - } - let index = u64::from_be_bytes([ - data[0], data[1], data[2], data[3], data[4], data[5], data[6], data[7], - ]); - Ok(OrderedMessage { - data: data[8..].to_vec(), - index, - }) - } -} - -/// Order messages by their index only, ignoring their data -impl PartialOrd for OrderedMessage { - fn partial_cmp(&self, other: &Self) -> Option { - Some((self.index).cmp(&(other.index))) - } -} - -#[cfg(test)] -mod ordered_message_to_bytes { - use super::*; - - #[test] - fn works() { - let message = OrderedMessage { - data: vec![123], - index: 1, - }; - let bytes = message.into_bytes(); - - let expected = vec![0, 0, 0, 0, 0, 0, 0, 1, 123]; - assert_eq!(expected, bytes); - } -} - -#[cfg(test)] -mod ordered_message_from_bytes { - use super::*; - - #[test] - fn fails_when_there_is_no_data() { - let result = OrderedMessage::try_from_bytes(Vec::new()); - assert_eq!(Err(MessageError::NoData), result); - } - - #[test] - fn fails_when_data_is_too_short() { - let result = OrderedMessage::try_from_bytes(vec![1, 2, 3]); - assert_eq!( - Err(MessageError::IndexTooShort { - received: 3, - expected: 8 - }), - result - ); - } - - #[test] - fn works_when_there_is_enough_to_make_a_sequence_number_but_no_message_data() { - let expected = OrderedMessage { - data: Vec::new(), - index: 1, - }; - let result = OrderedMessage::try_from_bytes(vec![0, 0, 0, 0, 0, 0, 0, 1]).unwrap(); - assert_eq!(expected, result); - } - - #[test] - fn works_when_there_is_seq_number_and_data() { - let expected = OrderedMessage { - data: vec![255, 255, 255], - index: 1, - }; - let result = - OrderedMessage::try_from_bytes(vec![0, 0, 0, 0, 0, 0, 0, 1, 255, 255, 255]).unwrap(); - assert_eq!(expected, result); - } -} - -#[test] -fn empty_message_does_not_affect_ordering() { - let mut msg1 = OrderedMessage { - data: vec![255, 255, 255], - index: 1, - }; - - let mut msg2 = OrderedMessage { - data: vec![], - index: 2, - }; - - assert!(msg1 < msg2); - - msg1.index = 2; - msg2.index = 1; - - assert!(msg1 > msg2); -} diff --git a/common/socks5/ordered-buffer/src/sender.rs b/common/socks5/ordered-buffer/src/sender.rs deleted file mode 100644 index f348b4e904..0000000000 --- a/common/socks5/ordered-buffer/src/sender.rs +++ /dev/null @@ -1,57 +0,0 @@ -use crate::message::OrderedMessage; - -/// Assigns sequence numbers to outbound byte vectors. These messages can then -/// be reassembled into an ordered sequence by the `OrderedMessageSender`. -#[derive(Debug)] -pub struct OrderedMessageSender { - next_index: u64, -} - -impl OrderedMessageSender { - pub fn new() -> OrderedMessageSender { - OrderedMessageSender { next_index: 0 } - } - - /// Turns raw bytes into an OrderedMessage containing the original bytes - /// and a sequence number; - pub fn wrap_message(&mut self, input: Vec) -> OrderedMessage { - let message = OrderedMessage { - data: input.to_vec(), - index: self.next_index, - }; - self.next_index += 1; - message - } -} - -impl Default for OrderedMessageSender { - fn default() -> Self { - OrderedMessageSender::new() - } -} - -#[cfg(test)] -mod ordered_message_sender { - use super::*; - - mod when_input_bytes_are_empty {} - - #[cfg(test)] - mod sequence_index_numbers { - use super::*; - - #[test] - fn increase_as_messages_are_sent() { - let mut sender = OrderedMessageSender::new(); - let first_bytes = vec![1, 2, 3, 4]; - let second_bytes = vec![5, 6, 7, 8]; - - let first_message = sender.wrap_message(first_bytes); - - assert_eq!(first_message.index, 0); - - let second_message = sender.wrap_message(second_bytes); - assert_eq!(second_message.index, 1); - } - } -} diff --git a/common/socks5/proxy-helpers/src/connection_controller.rs b/common/socks5/proxy-helpers/src/connection_controller.rs index a466316bbb..0d93009626 100644 --- a/common/socks5/proxy-helpers/src/connection_controller.rs +++ b/common/socks5/proxy-helpers/src/connection_controller.rs @@ -1,11 +1,11 @@ -// Copyright 2021 - Nym Technologies SA +// Copyright 2021-2023 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 use futures::channel::mpsc; use futures::StreamExt; use log::*; -use nym_ordered_buffer::{OrderedMessage, OrderedMessageBuffer, ReadContiguousData}; -use nym_socks5_requests::{ConnectionId, NetworkData, SendRequest}; +use nym_ordered_buffer::{OrderedMessageBuffer, ReadContiguousData}; +use nym_socks5_requests::{ConnectionId, SocketData}; use nym_task::connections::{ConnectionCommand, ConnectionCommandSender}; use nym_task::TaskClient; use std::collections::{HashMap, HashSet}; @@ -40,29 +40,13 @@ pub enum ControllerCommand { connection_id: ConnectionId, }, Send { - connection_id: ConnectionId, - data: Vec, - is_closed: bool, + data: SocketData, }, } -impl From for ControllerCommand { - fn from(value: NetworkData) -> Self { - ControllerCommand::Send { - connection_id: value.connection_id, - data: value.data, - is_closed: value.is_closed, - } - } -} - -impl From for ControllerCommand { - fn from(value: SendRequest) -> Self { - ControllerCommand::Send { - connection_id: value.conn_id, - data: value.data, - is_closed: value.local_closed, - } +impl ControllerCommand { + pub fn new_send(data: SocketData) -> Self { + ControllerCommand::Send { data } } } @@ -74,18 +58,13 @@ struct ActiveConnection { } impl ActiveConnection { - fn write_to_buf(&mut self, payload: Vec, is_closed: bool) { - let ordered_message = match OrderedMessage::try_from_bytes(payload) { - Ok(msg) => msg, - Err(err) => { - error!("Malformed ordered message - {err}"); - return; - } - }; + fn write_to_buf(&mut self, seq: u64, payload: Vec, is_closed: bool) { if is_closed { - self.closed_at_index = Some(ordered_message.index); + self.closed_at_index = Some(seq); + } + if let Err(err) = self.ordered_buffer.write(seq, payload) { + error!("failed to write to the buffer: {err}") } - self.ordered_buffer.write(ordered_message); } fn read_from_buf(&mut self) -> Option { @@ -117,7 +96,7 @@ pub struct Controller { // buffer for messages received before connection was established due to mixnet being able to // un-order messages. Note we don't ever expect to have more than 1-2 messages per connection here - pending_messages: HashMap, bool)>>, + pending_messages: HashMap>, shutdown: TaskClient, } @@ -154,8 +133,8 @@ impl Controller { // check if there were any pending messages if let Some(pending) = self.pending_messages.remove(&conn_id) { debug!("There were some pending messages for {}", conn_id); - for (payload, is_closed) in pending { - self.send_to_connection(conn_id, payload, is_closed) + for data in pending { + self.send_to_connection(data) } } } @@ -184,20 +163,25 @@ impl Controller { } } - fn send_to_connection(&mut self, conn_id: ConnectionId, payload: Vec, is_closed: bool) { - if let Some(active_connection) = self.active_connections.get_mut(&conn_id) { - if !payload.is_empty() { - active_connection.write_to_buf(payload, is_closed); - } else if !is_closed { - error!("Tried to write an empty message to a not-closing connection. Please let us know if you see this message"); - } + fn send_to_connection(&mut self, message: SocketData) { + let hdr = message.header; + if let Some(active_connection) = self.active_connections.get_mut(&hdr.connection_id) { + // always write to the buffer even if payload is empty (because it could have been the keep-alive message) + active_connection.write_to_buf(hdr.seq, message.data, hdr.local_socket_closed); if let Some(payload) = active_connection.read_from_buf() { if let Some(closed_at_index) = active_connection.closed_at_index { - if payload.last_index > closed_at_index { + if payload.last_sequence > closed_at_index { active_connection.is_closed = true; } } + + // however, don't send empty payload to the actual connection if it's not a close message + // TODO: or should we? + if payload.data.is_empty() && !active_connection.is_closed { + return; + } + if let Err(err) = active_connection .connection_sender .as_mut() @@ -207,34 +191,26 @@ impl Controller { socket_closed: active_connection.is_closed, }) { - error!("WTF IS THIS: {err}"); + error!("failed to send on the active connection channel: {err}"); } - - // TODO: ABOVE UNWRAP CAUSED A CRASH IN A NORMAL USE!!!! - // TODO: - // TODO: surprisingly it only happened on socks client, never on nSP - // TODO: - // TODO: - // TODO: - // TODO: } - } else if !self.recently_closed.contains(&conn_id) { + } else if !self.recently_closed.contains(&hdr.connection_id) { debug!("Received a 'Send' before 'Connect' - going to buffer the data"); let pending = self .pending_messages - .entry(conn_id) + .entry(hdr.connection_id) .or_insert_with(Vec::new); - pending.push((payload, is_closed)); - } else if !is_closed { + pending.push(message); + } else if !hdr.local_socket_closed { error!( "Tried to write to closed connection {} ({} bytes were 'lost)", - conn_id, - payload.len() + hdr.connection_id, + message.data.len() ); } else { debug!( "Tried to write to closed connection {}, but remote is already closed", - conn_id + hdr.connection_id ) } } @@ -243,8 +219,8 @@ impl Controller { loop { tokio::select! { command = self.receiver.next() => match command { - Some(ControllerCommand::Send{connection_id, data, is_closed}) => { - self.send_to_connection(connection_id, data, is_closed) + Some(ControllerCommand::Send{data}) => { + self.send_to_connection(data) } Some(ControllerCommand::Insert{connection_id, connection_sender}) => { self.insert_connection(connection_id, connection_sender) diff --git a/common/socks5/proxy-helpers/src/lib.rs b/common/socks5/proxy-helpers/src/lib.rs index cce784c4da..673dd9ea46 100644 --- a/common/socks5/proxy-helpers/src/lib.rs +++ b/common/socks5/proxy-helpers/src/lib.rs @@ -3,4 +3,5 @@ pub mod available_reader; pub mod connection_controller; +pub mod ordered_sender; pub mod proxy_runner; diff --git a/common/socks5/proxy-helpers/src/ordered_sender.rs b/common/socks5/proxy-helpers/src/ordered_sender.rs new file mode 100644 index 0000000000..d71f843558 --- /dev/null +++ b/common/socks5/proxy-helpers/src/ordered_sender.rs @@ -0,0 +1,116 @@ +// Copyright 2023 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::proxy_runner::MixProxySender; +use bytes::Bytes; +use log::{debug, error}; +use nym_socks5_requests::{ConnectionId, SocketData}; +use std::io; + +pub(crate) struct OrderedMessageSender { + connection_id: ConnectionId, + // addresses are provided for better logging + local_destination_address: String, + remote_source_address: String, + mixnet_sender: MixProxySender, + + next_message_seq: u64, + mix_message_adapter: F, +} + +impl OrderedMessageSender +where + F: Fn(SocketData) -> S, +{ + pub(crate) fn new( + local_destination_address: String, + remote_source_address: String, + connection_id: ConnectionId, + mixnet_sender: MixProxySender, + mix_message_adapter: F, + ) -> Self { + OrderedMessageSender { + local_destination_address, + remote_source_address, + connection_id, + mixnet_sender, + next_message_seq: 0, + mix_message_adapter, + } + } + + fn sequence(&mut self) -> u64 { + let next = self.next_message_seq; + self.next_message_seq += 1; + next + } + + fn construct_message(&mut self, data: Vec, local_socket_closed: bool) -> S { + let data = SocketData::new( + self.sequence(), + self.connection_id, + local_socket_closed, + data, + ); + (self.mix_message_adapter)(data) + } + + async fn send_message(&self, message: S) { + if self.mixnet_sender.send(message).await.is_err() { + panic!("BatchRealMessageReceiver has stopped receiving!") + } + } + + pub(crate) async fn send_empty_close(&mut self) { + let message = self.construct_message(Vec::new(), true); + self.send_message(message).await + } + + pub(crate) async fn send_empty_keepalive(&mut self) { + log::trace!("Sending keepalive for connection: {}", self.connection_id); + let message = self.construct_message(Vec::new(), false); + self.send_message(message).await + } + + pub(crate) fn process_data(&self, read_data: Option>) -> ProcessedData { + let (read_data, is_finished) = match read_data { + Some(data) => match data { + Ok(data) => (data, false), + Err(err) => { + error!(target: &*format!("({}) socks5 inbound", self.connection_id), "failed to read request from the socket - {err}"); + (Default::default(), true) + } + }, + None => (Default::default(), true), + }; + + ProcessedData { + data: read_data, + is_done: is_finished, + } + } + + fn log_sent_message(&self, data: &ProcessedData) { + debug!( + target: &*format!("({}) socks5 inbound", self.connection_id), + "[{} bytes]\t{} → local → mixnet → remote → {}. Local closed: {}", + data.data.len(), + self.local_destination_address, + self.remote_source_address, + data.is_done + ); + } + + /// Send data read from local socket into the mixnet + pub(crate) async fn send_data(&mut self, data: ProcessedData) { + self.log_sent_message(&data); + let message = self.construct_message(data.data.into(), data.is_done); + self.send_message(message).await; + } +} + +// helper wrapper to keep track of field meanings +pub(crate) struct ProcessedData { + data: Bytes, + pub(crate) is_done: bool, +} diff --git a/common/socks5/proxy-helpers/src/proxy_runner/inbound.rs b/common/socks5/proxy-helpers/src/proxy_runner/inbound.rs index 6ca2d7da46..82e6396a5f 100644 --- a/common/socks5/proxy-helpers/src/proxy_runner/inbound.rs +++ b/common/socks5/proxy-helpers/src/proxy_runner/inbound.rs @@ -1,106 +1,22 @@ -// Copyright 2021 - Nym Technologies SA +// Copyright 2021-2023 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -use super::MixProxySender; use super::SHUTDOWN_TIMEOUT; use crate::available_reader::AvailableReader; +use crate::ordered_sender::OrderedMessageSender; use crate::proxy_runner::KEEPALIVE_INTERVAL; -use bytes::Bytes; use futures::FutureExt; use futures::StreamExt; use log::*; -use nym_ordered_buffer::OrderedMessageSender; -use nym_socks5_requests::ConnectionId; +use nym_socks5_requests::{ConnectionId, SocketData}; use nym_task::connections::LaneQueueLengths; use nym_task::connections::TransmissionLane; use nym_task::TaskClient; -use std::fmt::Debug; +use std::sync::Arc; use std::time::Duration; -use std::{io, sync::Arc}; use tokio::select; use tokio::{net::tcp::OwnedReadHalf, sync::Notify, time::sleep}; -async fn send_empty_close( - connection_id: ConnectionId, - message_sender: &mut OrderedMessageSender, - mix_sender: &MixProxySender, - adapter_fn: F, -) where - F: Fn(ConnectionId, Vec, bool) -> S, - S: Debug, -{ - let ordered_msg = message_sender.wrap_message(Vec::new()).into_bytes(); - mix_sender - .send(adapter_fn(connection_id, ordered_msg, true)) - .await - .expect("BatchRealMessageReceiver has stopped receiving!"); -} - -async fn send_empty_keepalive( - connection_id: ConnectionId, - message_sender: &mut OrderedMessageSender, - mix_sender: &MixProxySender, - adapter_fn: F, -) where - F: Fn(ConnectionId, Vec, bool) -> S, - S: Debug, -{ - log::trace!("Sending keepalive for connection: {connection_id}"); - let ordered_msg = message_sender.wrap_message(Vec::new()).into_bytes(); - mix_sender - .send(adapter_fn(connection_id, ordered_msg, false)) - .await - .expect("BatchRealMessageReceiver has stopped receiving!"); -} - -async fn deal_with_data( - read_data: Option>, - local_destination_address: &str, - remote_source_address: &str, - connection_id: ConnectionId, - message_sender: &mut OrderedMessageSender, - mix_sender: &MixProxySender, - adapter_fn: F, -) -> bool -where - F: Fn(ConnectionId, Vec, bool) -> S, - S: Debug, -{ - let (read_data, is_finished) = match read_data { - Some(data) => match data { - Ok(data) => (data, false), - Err(err) => { - error!(target: &*format!("({connection_id}) socks5 inbound"), "failed to read request from the socket - {err}"); - (Default::default(), true) - } - }, - None => (Default::default(), true), - }; - - debug!( - target: &*format!("({connection_id}) socks5 inbound"), - "[{} bytes]\t{} → local → mixnet → remote → {}. Local closed: {}", - read_data.len(), - local_destination_address, - remote_source_address, - is_finished - ); - - // if we're sending through the mixnet increase the sequence number... - let ordered_msg = message_sender.wrap_message(read_data.to_vec()).into_bytes(); - log::trace!( - "pushing data down the input sender: size: {}", - ordered_msg.len() - ); - - mix_sender - .send(adapter_fn(connection_id, ordered_msg, is_finished)) - .await - .expect("InputMessageReceiver has stopped receiving!"); - - is_finished -} - async fn wait_until_lane_empty(lane_queue_lengths: &Option, connection_id: u64) { if let Some(lane_queue_lengths) = lane_queue_lengths { if tokio::time::timeout( @@ -158,27 +74,21 @@ async fn wait_for_lane( } } -#[allow(clippy::too_many_arguments)] pub(super) async fn run_inbound( mut reader: OwnedReadHalf, - local_destination_address: String, // addresses are provided for better logging - remote_source_address: String, + mut message_sender: OrderedMessageSender, connection_id: ConnectionId, - mix_sender: MixProxySender, available_plaintext_per_mix_packet: usize, - adapter_fn: F, shutdown_notify: Arc, lane_queue_lengths: Option, mut shutdown_listener: TaskClient, ) -> OwnedReadHalf where - F: Fn(ConnectionId, Vec, bool) -> S + Send + 'static, - S: Debug, + F: Fn(SocketData) -> S + Send + 'static, { // TODO: this multiplication by 4 is completely arbitrary here let mut available_reader = AvailableReader::new(&mut reader, Some(available_plaintext_per_mix_packet * 4)); - let mut message_sender = OrderedMessageSender::new(); // Shutdown if outbound signled to shutdown let shutdown_future = shutdown_notify.notified().then(|_| sleep(SHUTDOWN_TIMEOUT)); @@ -217,7 +127,7 @@ where ); // inform remote just in case it was closed because of lack of heartbeat. // worst case the remote will just have couple of false negatives - send_empty_close(connection_id, &mut message_sender, &mix_sender, &adapter_fn).await; + message_sender.send_empty_close().await; break; } _ = shutdown_listener.recv() => { @@ -233,7 +143,7 @@ where break; } _ = keepalive_timer.tick() => { - send_empty_keepalive(connection_id, &mut message_sender, &mix_sender, &adapter_fn).await; + message_sender.send_empty_keepalive().await; } // Read the next data when there is space in the lane. // The purpose of chaining the wait here is that it makes sure we can cancel the @@ -241,15 +151,12 @@ where read_data = wait_until_lane_almost_empty(&lane_queue_lengths, connection_id) .then(|_| available_reader.next()), if !we_are_closed => { - if deal_with_data( - read_data, - &local_destination_address, - &remote_source_address, - connection_id, - &mut message_sender, - &mix_sender, - &adapter_fn, - ).await { + let processed = message_sender.process_data(read_data); + let is_done = processed.is_done; + + message_sender.send_data(processed).await; + + if is_done { // After reading the last data, notify the closing_future to wait until the // lane is clear before exiting. // We don't wait here since we want to be able to cancel the wait on close or diff --git a/common/socks5/proxy-helpers/src/proxy_runner/mod.rs b/common/socks5/proxy-helpers/src/proxy_runner/mod.rs index 65e24938a8..38499ea59c 100644 --- a/common/socks5/proxy-helpers/src/proxy_runner/mod.rs +++ b/common/socks5/proxy-helpers/src/proxy_runner/mod.rs @@ -2,7 +2,8 @@ // SPDX-License-Identifier: Apache-2.0 use crate::connection_controller::ConnectionReceiver; -use nym_socks5_requests::ConnectionId; +use crate::ordered_sender::OrderedMessageSender; +use nym_socks5_requests::{ConnectionId, SocketData}; use nym_task::connections::LaneQueueLengths; use nym_task::TaskClient; use std::fmt::Debug; @@ -92,20 +93,24 @@ where // request/response as required by entity running particular side of the proxy. pub async fn run(mut self, adapter_fn: F) -> Self where - F: Fn(ConnectionId, Vec, bool) -> S + Send + Sync + 'static, + F: Fn(SocketData) -> S + Send + Sync + 'static, { let (read_half, write_half) = self.socket.take().unwrap().into_split(); let shutdown_notify = Arc::new(Notify::new()); // should run until either inbound closes or is notified from outbound - let inbound_future = inbound::run_inbound( - read_half, + let ordered_sender = OrderedMessageSender::new( self.local_destination_address.clone(), self.remote_source_address.clone(), self.connection_id, self.mix_sender.clone(), - self.available_plaintext_per_mix_packet, adapter_fn, + ); + let inbound_future = inbound::run_inbound( + read_half, + ordered_sender, + self.connection_id, + self.available_plaintext_per_mix_packet, Arc::clone(&shutdown_notify), self.lane_queue_lengths.clone(), self.shutdown_listener.clone(), diff --git a/common/socks5/requests/src/lib.rs b/common/socks5/requests/src/lib.rs index d6cff2a5e3..58c23dcd9d 100644 --- a/common/socks5/requests/src/lib.rs +++ b/common/socks5/requests/src/lib.rs @@ -3,6 +3,7 @@ use nym_service_providers_common::interface; use nym_service_providers_common::interface::ServiceProviderMessagingError; +use std::mem; use thiserror::Error; pub use request::*; @@ -16,6 +17,185 @@ pub mod version; pub type Socks5ProviderRequest = interface::Request; pub type Socks5ProviderResponse = interface::Response; +#[derive(Debug, Error, PartialEq, Eq)] +#[error( + "didn't receive enough data to recover socket data. got {received}, but expected at least {expected}" +)] +pub struct InsufficientSocketDataError { + received: usize, + expected: usize, +} + +#[derive(Debug, Clone, Copy, Eq, PartialEq)] +pub struct SocketDataHeader { + pub seq: u64, + pub connection_id: ConnectionId, + pub local_socket_closed: bool, +} + +impl SocketDataHeader { + const SERIALIZED_LEN: usize = mem::size_of::() + 1 + mem::size_of::(); + + // we need to have two serialization methods for backwards compatibility, + // since we serialized those fields differently depending on whether it was ingress vs egress... + + pub fn try_from_request_bytes( + b: &[u8], + ) -> Result { + if b.len() != Self::SERIALIZED_LEN { + return Err(InsufficientSocketDataError { + received: b.len(), + expected: Self::SERIALIZED_LEN, + }); + } + + // the unwraps here are fine as we just ensured we have the exact amount of bytes we need + let connection_id = ConnectionId::from_be_bytes(b[0..8].try_into().unwrap()); + let local_socket_closed = b[8] != 0; + let seq = u64::from_be_bytes(b[9..].try_into().unwrap()); + + Ok(SocketDataHeader { + seq, + connection_id, + local_socket_closed, + }) + } + + // the serialization of the header looks as follows: + // (it's vital it's not modified as we need this exact structure for backwards compatibility) + // CONNECTION_ID (8B) || SOCKET_CLOSED (1B) || SEQUENCE (8B) + pub fn into_request_bytes(self) -> Vec { + self.into_request_bytes_iter().collect() + } + + pub fn into_request_bytes_iter(self) -> impl Iterator { + self.connection_id + .to_be_bytes() + .into_iter() + .chain(std::iter::once(self.local_socket_closed as u8)) + .chain(self.seq.to_be_bytes().into_iter()) + } + + pub fn try_from_response_bytes( + b: &[u8], + ) -> Result { + if b.len() != Self::SERIALIZED_LEN { + return Err(InsufficientSocketDataError { + received: b.len(), + expected: Self::SERIALIZED_LEN, + }); + } + + // the unwraps here are fine as we just ensured we have the exact amount of bytes we need + let local_socket_closed = b[0] != 0; + let connection_id = ConnectionId::from_be_bytes(b[1..9].try_into().unwrap()); + let seq = u64::from_be_bytes(b[9..].try_into().unwrap()); + + Ok(SocketDataHeader { + seq, + connection_id, + local_socket_closed, + }) + } + + // the serialization of the header looks as follows: + // (it's vital it's not modified as we need this exact structure for backwards compatibility) + // SOCKET_CLOSED (1B) || CONNECTION_ID (8B) || SEQUENCE (8B) + pub fn into_response_bytes(self) -> Vec { + self.into_response_bytes_iter().collect() + } + + pub fn into_response_bytes_iter(self) -> impl Iterator { + std::iter::once(self.local_socket_closed as u8) + .chain(self.connection_id.to_be_bytes().into_iter()) + .chain(self.seq.to_be_bytes().into_iter()) + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct SocketData { + pub header: SocketDataHeader, + pub data: Vec, +} + +impl SocketData { + pub fn new( + seq: u64, + connection_id: ConnectionId, + local_socket_closed: bool, + data: Vec, + ) -> Self { + SocketData { + header: SocketDataHeader { + seq, + connection_id, + local_socket_closed, + }, + data, + } + } + + fn verify_deserialization_len(b: &[u8]) -> Result<(), InsufficientSocketDataError> { + if b.is_empty() { + return Err(InsufficientSocketDataError { + received: 0, + expected: SocketDataHeader::SERIALIZED_LEN, + }); + } + + if b.len() < SocketDataHeader::SERIALIZED_LEN { + return Err(InsufficientSocketDataError { + received: b.len(), + expected: SocketDataHeader::SERIALIZED_LEN, + }); + } + Ok(()) + } + + // we need to have two serialization methods for backwards compatibility, + // since we serialized those fields differently depending on whether it was ingress vs egress... + pub fn try_from_request_bytes(b: &[u8]) -> Result { + Self::verify_deserialization_len(b)?; + let header = + SocketDataHeader::try_from_request_bytes(&b[..SocketDataHeader::SERIALIZED_LEN])?; + let data = b[SocketDataHeader::SERIALIZED_LEN..].to_vec(); + + Ok(SocketData { header, data }) + } + + // the serialization of the socket data looks as follows: + // HEADER || DATA + pub fn into_request_bytes(self) -> Vec { + self.into_request_bytes_iter().collect() + } + + pub fn into_request_bytes_iter(self) -> impl Iterator { + self.header + .into_request_bytes_iter() + .chain(self.data.into_iter()) + } + + pub fn try_from_response_bytes(b: &[u8]) -> Result { + Self::verify_deserialization_len(b)?; + + let header = + SocketDataHeader::try_from_response_bytes(&b[..SocketDataHeader::SERIALIZED_LEN])?; + let data = b[SocketDataHeader::SERIALIZED_LEN..].to_vec(); + + Ok(SocketData { header, data }) + } + + pub fn into_response_bytes(self) -> Vec { + self.into_response_bytes_iter().collect() + } + + pub fn into_response_bytes_iter(self) -> impl Iterator { + self.header + .into_response_bytes_iter() + .chain(self.data.into_iter()) + } +} + #[derive(Debug, Error)] pub enum Socks5RequestError { #[error("failed to deserialize received request: {source}")] @@ -51,6 +231,103 @@ mod tests { use super::*; use nym_service_providers_common::interface::RequestContent; + #[cfg(test)] + mod socket_data_serialization { + use super::*; + + #[test] + fn for_requests() { + assert_eq!( + InsufficientSocketDataError { + received: 0, + expected: SocketDataHeader::SERIALIZED_LEN + }, + SocketData::try_from_request_bytes(&[]).unwrap_err() + ); + + assert_eq!( + InsufficientSocketDataError { + received: 10, + expected: SocketDataHeader::SERIALIZED_LEN + }, + SocketData::try_from_request_bytes(&[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]).unwrap_err() + ); + + let good_data = SocketData::new(42, 12345, false, vec![2, 3]); + let serialized = good_data.clone().into_request_bytes(); + + assert_eq!( + good_data, + SocketData::try_from_request_bytes(&serialized).unwrap() + ); + assert_ne!( + good_data, + SocketData::try_from_response_bytes(&serialized).unwrap() + ); + + let raw_bytes = [ + 6, 6, 6, 6, 6, 6, 6, 6, 0, 0, 1, 2, 3, 4, 5, 6, 7, 255, 255, 255, + ]; + assert_eq!( + SocketData { + header: SocketDataHeader { + seq: u64::from_be_bytes([0, 1, 2, 3, 4, 5, 6, 7]), + connection_id: ConnectionId::from_be_bytes([6, 6, 6, 6, 6, 6, 6, 6]), + local_socket_closed: false, + }, + data: vec![255, 255, 255], + }, + SocketData::try_from_request_bytes(&raw_bytes).unwrap() + ) + } + + #[test] + fn for_responses() { + assert_eq!( + InsufficientSocketDataError { + received: 0, + expected: SocketDataHeader::SERIALIZED_LEN + }, + SocketData::try_from_response_bytes(&[]).unwrap_err() + ); + + assert_eq!( + InsufficientSocketDataError { + received: 10, + expected: SocketDataHeader::SERIALIZED_LEN + }, + SocketData::try_from_response_bytes(&[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]).unwrap_err() + ); + + let good_data = SocketData::new(42, 12345, false, vec![2, 3]); + let serialized = good_data.clone().into_response_bytes(); + + assert_eq!( + good_data, + SocketData::try_from_response_bytes(&serialized).unwrap() + ); + assert_ne!( + good_data, + SocketData::try_from_request_bytes(&serialized).unwrap() + ); + + let raw_bytes = [ + 0, 6, 6, 6, 6, 6, 6, 6, 6, 0, 1, 2, 3, 4, 5, 6, 7, 255, 255, 255, + ]; + assert_eq!( + SocketData { + header: SocketDataHeader { + seq: u64::from_be_bytes([0, 1, 2, 3, 4, 5, 6, 7]), + connection_id: ConnectionId::from_be_bytes([6, 6, 6, 6, 6, 6, 6, 6]), + local_socket_closed: false, + }, + data: vec![255, 255, 255], + }, + SocketData::try_from_response_bytes(&raw_bytes).unwrap() + ) + } + } + #[cfg(test)] mod interface_backwards_compatibility { use super::*; @@ -99,9 +376,10 @@ mod tests { match new_deserialized.content { RequestContent::ProviderData(req) => match req.content { Socks5RequestContent::Send(send_req) => { - assert_eq!(send_req.conn_id, 7810961472501196273); - assert_eq!(send_req.data.len(), 111); - assert!(!send_req.local_closed); + assert_eq!(send_req.data.header.connection_id, 7810961472501196273); + assert_eq!(send_req.data.header.seq, 0); + assert_eq!(send_req.data.data.len(), 103); + assert!(!send_req.data.header.local_socket_closed); } _ => panic!("unexpected request"), }, diff --git a/common/socks5/requests/src/request.rs b/common/socks5/requests/src/request.rs index 78e07e4f02..0e29fb49d9 100644 --- a/common/socks5/requests/src/request.rs +++ b/common/socks5/requests/src/request.rs @@ -1,7 +1,10 @@ // Copyright 2020-2023 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -use crate::{make_bincode_serializer, Socks5ProtocolVersion, Socks5RequestError, Socks5Response}; +use crate::{ + make_bincode_serializer, InsufficientSocketDataError, SocketData, Socks5ProtocolVersion, + Socks5RequestError, Socks5Response, +}; use nym_service_providers_common::interface::{Serializable, ServiceProviderRequest}; use nym_sphinx_addressing::clients::{Recipient, RecipientFormattingError}; use serde::{Deserialize, Serialize}; @@ -61,6 +64,9 @@ pub enum RequestDeserializationError { #[from] source: bincode::Error, }, + + #[error(transparent)] + InvalidSocketData(#[from] InsufficientSocketDataError), } impl RequestDeserializationError { @@ -79,9 +85,7 @@ pub struct ConnectRequest { #[derive(Debug, Clone, PartialEq, Eq)] pub struct SendRequest { - pub conn_id: ConnectionId, - pub data: Vec, - pub local_closed: bool, + pub data: SocketData, } #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] @@ -178,15 +182,10 @@ impl Socks5Request { } } - pub fn new_send( - protocol_version: Socks5ProtocolVersion, - conn_id: ConnectionId, - data: Vec, - local_closed: bool, - ) -> Socks5Request { + pub fn new_send(protocol_version: Socks5ProtocolVersion, data: SocketData) -> Socks5Request { Socks5Request { protocol_version, - content: Socks5RequestContent::new_send(conn_id, data, local_closed), + content: Socks5RequestContent::new_send(data), } } @@ -231,16 +230,8 @@ impl Socks5RequestContent { } /// Construct a new Request::Send instance - pub fn new_send( - conn_id: ConnectionId, - data: Vec, - local_closed: bool, - ) -> Socks5RequestContent { - Socks5RequestContent::Send(SendRequest { - conn_id, - data, - local_closed, - }) + pub fn new_send(data: SocketData) -> Socks5RequestContent { + Socks5RequestContent::Send(SendRequest { data }) } /// Deserialize the request type, connection id, destination address and port, @@ -257,6 +248,14 @@ impl Socks5RequestContent { /// The request_flag tells us whether this is a new connection request (`new_connect`), /// an already-established connection we should send up (`new_send`), or /// a request to close an established connection (`new_close`). + + // connect: + // RequestFlag::Connect || CONN_ID || ADDR_LEN || ADDR || + // + // send: + // RequestFlag::Send || CONN_ID || LOCAL_CLOSED || DATA + // where DATA: SEQ || TRUE_DATA + pub fn try_from_bytes(b: &[u8]) -> Result { // each request needs to at least contain flag and ConnectionId if b.is_empty() { @@ -314,21 +313,9 @@ impl Socks5RequestContent { return_address, )) } - RequestFlag::Send => { - if b.len() < 9 { - return Err(RequestDeserializationError::ConnectionIdTooShort); - } - let conn_id = u64::from_be_bytes([b[1], b[2], b[3], b[4], b[5], b[6], b[7], b[8]]); - - let local_closed = b[9] != 0; - let data = b[10..].to_vec(); - - Ok(Socks5RequestContent::Send(SendRequest { - conn_id, - data, - local_closed, - })) - } + RequestFlag::Send => Ok(Socks5RequestContent::Send(SendRequest { + data: SocketData::try_from_request_bytes(&b[1..])?, + })), RequestFlag::Query => { use bincode::Options; let query = make_bincode_serializer().deserialize(&b[1..])?; @@ -359,9 +346,7 @@ impl Socks5RequestContent { } } Socks5RequestContent::Send(req) => std::iter::once(RequestFlag::Send as u8) - .chain(req.conn_id.to_be_bytes().into_iter()) - .chain(std::iter::once(req.local_closed as u8)) - .chain(req.data.into_iter()) + .chain(req.data.into_request_bytes_iter()) .collect(), Socks5RequestContent::Query(query) => { @@ -618,26 +603,7 @@ mod request_deserialization_tests { #[test] fn works_when_request_is_sized_properly_even_without_data() { - // correct 8 bytes of connection_id, 1 byte of local_closed and 0 bytes request data - let request_bytes = [RequestFlag::Send as u8, 1, 2, 3, 4, 5, 6, 7, 8, 0].to_vec(); - let request = Socks5RequestContent::try_from_bytes(&request_bytes).unwrap(); - match request { - Socks5RequestContent::Send(SendRequest { - conn_id, - data, - local_closed, - }) => { - assert_eq!(u64::from_be_bytes([1, 2, 3, 4, 5, 6, 7, 8]), conn_id); - assert_eq!(Vec::::new(), data); - assert!(!local_closed) - } - _ => unreachable!(), - } - } - - #[test] - fn works_when_request_is_sized_properly_and_has_data() { - // correct 8 bytes of connection_id, 1 byte of local_closed and 3 bytes request data (all 255) + // correct 8 bytes of connection_id, 1 byte of local_closed, 8 bytes of sequence and 0 bytes request data let request_bytes = [ RequestFlag::Send as u8, 1, @@ -649,6 +615,53 @@ mod request_deserialization_tests { 7, 8, 0, + 0, + 0, + 0, + 0, + 0, + 0, + 0, + 1, + ] + .to_vec(); + let request = Socks5RequestContent::try_from_bytes(&request_bytes).unwrap(); + match request { + Socks5RequestContent::Send(SendRequest { data }) => { + assert_eq!( + u64::from_be_bytes([1, 2, 3, 4, 5, 6, 7, 8]), + data.header.connection_id + ); + assert!(!data.header.local_socket_closed); + assert_eq!(1, data.header.seq); + assert!(data.data.is_empty()); + } + _ => unreachable!(), + } + } + + #[test] + fn works_when_request_is_sized_properly_and_has_data() { + // correct 8 bytes of connection_id, 1 byte of local_closed, 8 bytes of sequence and 3 bytes request data (all 255) + let request_bytes = [ + RequestFlag::Send as u8, + 1, + 2, + 3, + 4, + 5, + 6, + 7, + 8, + 1, + 0, + 0, + 0, + 0, + 0, + 0, + 0, + 1, 255, 255, 255, @@ -657,14 +670,14 @@ mod request_deserialization_tests { let request = Socks5RequestContent::try_from_bytes(&request_bytes).unwrap(); match request { - Socks5RequestContent::Send(SendRequest { - conn_id, - data, - local_closed, - }) => { - assert_eq!(u64::from_be_bytes([1, 2, 3, 4, 5, 6, 7, 8]), conn_id); - assert_eq!(vec![255, 255, 255], data); - assert!(!local_closed) + Socks5RequestContent::Send(SendRequest { data }) => { + assert_eq!( + u64::from_be_bytes([1, 2, 3, 4, 5, 6, 7, 8]), + data.header.connection_id + ); + assert!(data.header.local_socket_closed); + assert_eq!(1, data.header.seq); + assert_eq!(vec![255, 255, 255], data.data); } _ => unreachable!(), } diff --git a/common/socks5/requests/src/response.rs b/common/socks5/requests/src/response.rs index d0a34e074b..4e0c906439 100644 --- a/common/socks5/requests/src/response.rs +++ b/common/socks5/requests/src/response.rs @@ -1,7 +1,10 @@ // Copyright 2020-2023 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -use crate::{make_bincode_serializer, ConnectionId, Socks5ProtocolVersion, Socks5RequestError}; +use crate::{ + make_bincode_serializer, ConnectionId, InsufficientSocketDataError, SocketData, + Socks5ProtocolVersion, Socks5RequestError, +}; use nym_service_providers_common::interface::{Serializable, ServiceProviderResponse}; use serde::{Deserialize, Serialize}; use tap::TapFallible; @@ -33,6 +36,12 @@ impl TryFrom for ResponseFlag { #[derive(Debug, Error)] pub enum ResponseDeserializationError { + #[error("the network data was malformed: {source}")] + MalformedNetworkData { + #[from] + source: InsufficientSocketDataError, + }, + #[error("not enough bytes to recover the connection id")] ConnectionIdTooShort, @@ -115,23 +124,14 @@ impl Socks5Response { pub fn new_network_data( protocol_version: Socks5ProtocolVersion, + seq: u64, connection_id: ConnectionId, data: Vec, is_closed: bool, ) -> Socks5Response { Socks5Response { protocol_version, - content: Socks5ResponseContent::new_network_data(connection_id, data, is_closed), - } - } - - pub fn new_closed_empty( - protocol_version: Socks5ProtocolVersion, - connection_id: ConnectionId, - ) -> Socks5Response { - Socks5Response { - protocol_version, - content: Socks5ResponseContent::new_closed_empty(connection_id), + content: Socks5ResponseContent::new_network_data(seq, connection_id, data, is_closed), } } @@ -159,22 +159,21 @@ impl Socks5Response { #[derive(Clone, Debug, PartialEq, Eq)] pub enum Socks5ResponseContent { - NetworkData(NetworkData), + NetworkData { content: SocketData }, ConnectionError(ConnectionError), Query(QueryResponse), } impl Socks5ResponseContent { pub fn new_network_data( + seq: u64, connection_id: ConnectionId, data: Vec, is_closed: bool, ) -> Socks5ResponseContent { - Socks5ResponseContent::NetworkData(NetworkData::new(connection_id, data, is_closed)) - } - - pub fn new_closed_empty(connection_id: ConnectionId) -> Socks5ResponseContent { - Socks5ResponseContent::NetworkData(NetworkData::new_closed_empty(connection_id)) + Socks5ResponseContent::NetworkData { + content: SocketData::new(seq, connection_id, is_closed, data), + } } pub fn new_connection_error( @@ -186,9 +185,9 @@ impl Socks5ResponseContent { pub fn into_bytes(self) -> Vec { match self { - Socks5ResponseContent::NetworkData(res) => { + Socks5ResponseContent::NetworkData { content } => { std::iter::once(ResponseFlag::NetworkData as u8) - .chain(res.into_bytes().into_iter()) + .chain(content.into_response_bytes_iter()) .collect() } Socks5ResponseContent::ConnectionError(res) => { @@ -220,9 +219,9 @@ impl Socks5ResponseContent { let response_flag = ResponseFlag::try_from(b[0])?; match response_flag { - ResponseFlag::NetworkData => Ok(Socks5ResponseContent::NetworkData( - NetworkData::try_from_bytes(&b[1..])?, - )), + ResponseFlag::NetworkData => Ok(Socks5ResponseContent::NetworkData { + content: SocketData::try_from_response_bytes(&b[1..])?, + }), ResponseFlag::ConnectionError => Ok(Socks5ResponseContent::ConnectionError( ConnectionError::try_from_bytes(&b[1..])?, )), @@ -235,73 +234,6 @@ impl Socks5ResponseContent { } } -/// A remote network network data response retrieved by the Socks5 service provider. This -/// can be serialized and sent back through the mixnet to the requesting -/// application. -#[derive(Clone, Debug, PartialEq, Eq)] -pub struct NetworkData { - pub data: Vec, - pub connection_id: ConnectionId, - pub is_closed: bool, -} - -impl NetworkData { - /// Constructor for responses - pub fn new(connection_id: ConnectionId, data: Vec, is_closed: bool) -> Self { - NetworkData { - data, - connection_id, - is_closed, - } - } - - pub fn new_closed_empty(connection_id: ConnectionId) -> Self { - NetworkData { - data: vec![], - connection_id, - is_closed: false, - } - } - - pub fn try_from_bytes(b: &[u8]) -> Result { - if b.is_empty() { - return Err(ResponseDeserializationError::NoData); - } - - let is_closed = b[0] != 0; - - if b.len() < 9 { - return Err(ResponseDeserializationError::ConnectionIdTooShort); - } - - let mut connection_id_bytes = b.to_vec(); - let data = connection_id_bytes.split_off(9); - - let connection_id = u64::from_be_bytes([ - connection_id_bytes[1], - connection_id_bytes[2], - connection_id_bytes[3], - connection_id_bytes[4], - connection_id_bytes[5], - connection_id_bytes[6], - connection_id_bytes[7], - connection_id_bytes[8], - ]); - - let response = NetworkData::new(connection_id, data, is_closed); - Ok(response) - } - - /// Serializes the response into bytes so that it can be sent back through - /// the mixnet to the requesting application. - pub fn into_bytes(self) -> Vec { - std::iter::once(self.is_closed as u8) - .chain(self.connection_id.to_be_bytes().iter().cloned()) - .chain(self.data.into_iter()) - .collect() - } -} - #[derive(Clone, Debug, PartialEq, Eq)] pub struct ConnectionError { pub connection_id: ConnectionId, @@ -366,57 +298,6 @@ pub enum QueryResponse { mod tests { use super::*; - #[cfg(test)] - mod constructing_socks5_data_responses_from_bytes { - use super::*; - - #[test] - fn fails_when_zero_bytes_are_supplied() { - let response_bytes = Vec::new(); - - let err = NetworkData::try_from_bytes(&response_bytes).unwrap_err(); - assert!(matches!(err, ResponseDeserializationError::NoData)); - } - - #[test] - fn fails_when_connection_id_bytes_are_too_short() { - let response_bytes = vec![0, 1, 2, 3, 4, 5, 6]; - let err = NetworkData::try_from_bytes(&response_bytes).unwrap_err(); - assert!(matches!( - err, - ResponseDeserializationError::ConnectionIdTooShort, - )); - } - - #[test] - fn works_when_there_is_no_data() { - let response_bytes = vec![0, 0, 1, 2, 3, 4, 5, 6, 7]; - let expected = NetworkData::new( - u64::from_be_bytes([0, 1, 2, 3, 4, 5, 6, 7]), - Vec::new(), - false, - ); - let actual = NetworkData::try_from_bytes(&response_bytes).unwrap(); - assert_eq!(expected.connection_id, actual.connection_id); - assert_eq!(expected.data, actual.data); - assert_eq!(expected.is_closed, actual.is_closed); - } - - #[test] - fn works_when_there_is_data() { - let response_bytes = vec![0, 0, 1, 2, 3, 4, 5, 6, 7, 255, 255, 255]; - let expected = NetworkData::new( - u64::from_be_bytes([0, 1, 2, 3, 4, 5, 6, 7]), - vec![255, 255, 255], - false, - ); - let actual = NetworkData::try_from_bytes(&response_bytes).unwrap(); - assert_eq!(expected.connection_id, actual.connection_id); - assert_eq!(expected.data, actual.data); - assert_eq!(expected.is_closed, actual.is_closed); - } - } - #[cfg(test)] mod connection_error_response_serde_tests { use super::*; diff --git a/service-providers/network-requester/Cargo.toml b/service-providers/network-requester/Cargo.toml index 0333d90523..32e8c18058 100644 --- a/service-providers/network-requester/Cargo.toml +++ b/service-providers/network-requester/Cargo.toml @@ -43,7 +43,6 @@ nym-bin-common = { path = "../../common/bin-common", features = ["output_format" nym-network-defaults = { path = "../../common/network-defaults" } nym-sdk = { path = "../../sdk/rust/nym-sdk" } nym-sphinx = { path = "../../common/nymsphinx" } -nym-ordered-buffer = {path = "../../common/socks5/ordered-buffer"} nym-socks5-proxy-helpers = { path = "../../common/socks5/proxy-helpers" } nym-service-providers-common = { path = "../common" } nym-socks5-requests = { path = "../../common/socks5/requests" } diff --git a/service-providers/network-requester/src/core.rs b/service-providers/network-requester/src/core.rs index d948c79c6d..7e4438c2ce 100644 --- a/service-providers/network-requester/src/core.rs +++ b/service-providers/network-requester/src/core.rs @@ -24,7 +24,7 @@ use nym_socks5_proxy_helpers::connection_controller::{ }; use nym_socks5_proxy_helpers::proxy_runner::{MixProxyReader, MixProxySender}; use nym_socks5_requests::{ - ConnectRequest, ConnectionId, NetworkData, QueryRequest, QueryResponse, SendRequest, + ConnectRequest, ConnectionId, QueryRequest, QueryResponse, SendRequest, SocketData, Socks5ProtocolVersion, Socks5ProviderRequest, Socks5Request, Socks5RequestContent, Socks5Response, }; @@ -137,13 +137,13 @@ impl ServiceProvider for NRServiceProvider { .connected_services .read() .await - .get(&req.conn_id) + .get(&req.data.header.connection_id) { stats_collector .request_stats_data .write() .await - .processed(remote_addr, req.data.len() as u32); + .processed(remote_addr, req.data.data.len() as u32); } } self.handle_proxy_send(req) @@ -359,7 +359,7 @@ impl NRServiceProvider { return_address, remote_version, connection_id, - NetworkData::new_closed_empty(connection_id), + SocketData::new(0, connection_id, true, Vec::new()), ); mix_input_sender @@ -471,7 +471,9 @@ impl NRServiceProvider { } fn handle_proxy_send(&mut self, req: SendRequest) { - self.controller_sender.unbounded_send(req.into()).unwrap() + self.controller_sender + .unbounded_send(ControllerCommand::new_send(req.data)) + .unwrap() } fn handle_query( diff --git a/service-providers/network-requester/src/reply.rs b/service-providers/network-requester/src/reply.rs index 6fe313e30b..7aba7473d1 100644 --- a/service-providers/network-requester/src/reply.rs +++ b/service-providers/network-requester/src/reply.rs @@ -6,7 +6,7 @@ use nym_service_providers_common::interface::{ ControlRequest, ControlResponse, ProviderInterfaceVersion, RequestVersion, }; use nym_socks5_requests::{ - ConnectionId, NetworkData, Socks5ProviderRequest, Socks5ProviderResponse, Socks5Request, + ConnectionId, SocketData, Socks5ProviderRequest, Socks5ProviderResponse, Socks5Request, Socks5RequestContent, Socks5Response, Socks5ResponseContent, }; use nym_sphinx::addressing::clients::Recipient; @@ -78,12 +78,12 @@ impl MixnetMessage { address: MixnetAddress, request_version: RequestVersion, connection_id: ConnectionId, - content: NetworkData, + content: SocketData, ) -> Self { // TODO: simplify by providing better constructor for `PlaceholderResponse` let res = Socks5Response::new( request_version.provider_protocol, - Socks5ResponseContent::NetworkData(content), + Socks5ResponseContent::NetworkData { content }, ); let msg = Socks5ProviderResponse::new_provider_data(request_version.provider_interface, res); @@ -135,11 +135,15 @@ impl MixnetMessage { pub(crate) fn new_network_data_response_content( address: MixnetAddress, request_version: RequestVersion, + seq: u64, connection_id: ConnectionId, data: Vec, closed_socket: bool, ) -> Self { - let response_content = NetworkData::new(connection_id, data, closed_socket); + if seq == 0 && data.is_empty() { + println!("new empty response with 0 seq") + } + let response_content = SocketData::new(seq, connection_id, closed_socket, data); Self::new_network_data_response(address, request_version, connection_id, response_content) } diff --git a/service-providers/network-requester/src/socks5/tcp.rs b/service-providers/network-requester/src/socks5/tcp.rs index 2a50bbf3a3..baed05393d 100644 --- a/service-providers/network-requester/src/socks5/tcp.rs +++ b/service-providers/network-requester/src/socks5/tcp.rs @@ -66,13 +66,14 @@ impl Connection { Some(lane_queue_lengths), shutdown, ) - .run(move |conn_id, read_data, socket_closed| { + .run(move |socket_data| { MixnetMessage::new_network_data_response_content( return_address.clone(), remote_version.clone(), - conn_id, - read_data, - socket_closed, + socket_data.header.seq, + socket_data.header.connection_id, + socket_data.data, + socket_data.header.local_socket_closed, ) }) .await diff --git a/service-providers/network-requester/src/statistics/collector.rs b/service-providers/network-requester/src/statistics/collector.rs index d728db8df6..e0169165fb 100644 --- a/service-providers/network-requester/src/statistics/collector.rs +++ b/service-providers/network-requester/src/statistics/collector.rs @@ -6,10 +6,11 @@ use crate::core::new_legacy_request_version; use crate::reply::MixnetMessage; use async_trait::async_trait; use log::*; -use nym_ordered_buffer::OrderedMessageSender; use nym_service_providers_common::interface::RequestVersion; use nym_socks5_proxy_helpers::proxy_runner::MixProxySender; -use nym_socks5_requests::{ConnectionId, RemoteAddress, Socks5Request, Socks5RequestContent}; +use nym_socks5_requests::{ + ConnectionId, RemoteAddress, SocketData, Socks5Request, Socks5RequestContent, +}; use nym_sphinx::addressing::clients::Recipient; use nym_statistics_common::api::{ build_statistics_request_bytes, DEFAULT_STATISTICS_SERVICE_ADDRESS, @@ -190,9 +191,9 @@ impl StatisticsCollector for ServiceStatisticsCollector { .expect("MixProxyReader has stopped receiving!"); trace!("Sending data to statistics service"); - let mut message_sender = OrderedMessageSender::new(); - let ordered_msg = message_sender.wrap_message(msg).into_bytes(); - let send_req = Socks5RequestContent::new_send(conn_id, ordered_msg, true); + + let message = SocketData::new(0, conn_id, true, msg); + let send_req = Socks5RequestContent::new_send(message); let mixnet_message = MixnetMessage::new_network_data_request( self.stats_provider_addr,