From f8c2f9050213189edb624301c43c8d4e77ffb401 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Wed, 7 Jun 2023 14:03:44 +0100 Subject: [PATCH 01/12] initial work on putting data sequence explicitly inside socks5 request --- common/socks5-client-core/src/socks/client.rs | 12 +- common/socks5/ordered-buffer/src/sender.rs | 40 +++--- .../src/connection_controller.rs | 8 +- .../proxy-helpers/src/proxy_runner/inbound.rs | 33 ++--- .../proxy-helpers/src/proxy_runner/mod.rs | 4 +- common/socks5/requests/src/lib.rs | 127 +++++++++++++++++- common/socks5/requests/src/request.rs | 63 ++++----- common/socks5/requests/src/response.rs | 1 + .../network-requester/src/core.rs | 4 +- .../network-requester/src/socks5/tcp.rs | 8 +- .../src/statistics/collector.rs | 9 +- 11 files changed, 210 insertions(+), 99 deletions(-) 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/ordered-buffer/src/sender.rs b/common/socks5/ordered-buffer/src/sender.rs index f348b4e904..40da98ef26 100644 --- a/common/socks5/ordered-buffer/src/sender.rs +++ b/common/socks5/ordered-buffer/src/sender.rs @@ -1,27 +1,31 @@ -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, + next_sequence: u64, } impl OrderedMessageSender { pub fn new() -> OrderedMessageSender { - OrderedMessageSender { next_index: 0 } + OrderedMessageSender { next_sequence: 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 + pub fn sequence(&mut self) -> u64 { + let next = self.next_sequence; + self.next_sequence += 1; + next } + + // /// 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_sequence, + // }; + // self.next_sequence += 1; + // message + // } } impl Default for OrderedMessageSender { @@ -43,15 +47,13 @@ mod ordered_message_sender { #[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); + let first_message = sender.sequence(); - assert_eq!(first_message.index, 0); + assert_eq!(first_message, 0); - let second_message = sender.wrap_message(second_bytes); - assert_eq!(second_message.index, 1); + let second_message = sender.sequence(); + assert_eq!(second_message, 1); } } } diff --git a/common/socks5/proxy-helpers/src/connection_controller.rs b/common/socks5/proxy-helpers/src/connection_controller.rs index a466316bbb..c1a9ff3f47 100644 --- a/common/socks5/proxy-helpers/src/connection_controller.rs +++ b/common/socks5/proxy-helpers/src/connection_controller.rs @@ -58,10 +58,12 @@ impl From for ControllerCommand { impl From for ControllerCommand { fn from(value: SendRequest) -> Self { + // TODO: do we care about sequence number here? + ControllerCommand::Send { - connection_id: value.conn_id, - data: value.data, - is_closed: value.local_closed, + connection_id: value.data.header.connection_id, + data: value.data.data, + is_closed: value.data.header.local_socket_closed, } } } diff --git a/common/socks5/proxy-helpers/src/proxy_runner/inbound.rs b/common/socks5/proxy-helpers/src/proxy_runner/inbound.rs index 6ca2d7da46..e6947d4c25 100644 --- a/common/socks5/proxy-helpers/src/proxy_runner/inbound.rs +++ b/common/socks5/proxy-helpers/src/proxy_runner/inbound.rs @@ -1,4 +1,4 @@ -// Copyright 2021 - Nym Technologies SA +// Copyright 2021-2023 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 use super::MixProxySender; @@ -10,7 +10,7 @@ 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; @@ -26,12 +26,12 @@ async fn send_empty_close( mix_sender: &MixProxySender, adapter_fn: F, ) where - F: Fn(ConnectionId, Vec, bool) -> S, + F: Fn(SocketData) -> S, S: Debug, { - let ordered_msg = message_sender.wrap_message(Vec::new()).into_bytes(); + let message = SocketData::new(message_sender.sequence(), connection_id, true, Vec::new()); mix_sender - .send(adapter_fn(connection_id, ordered_msg, true)) + .send(adapter_fn(message)) .await .expect("BatchRealMessageReceiver has stopped receiving!"); } @@ -42,13 +42,14 @@ async fn send_empty_keepalive( mix_sender: &MixProxySender, adapter_fn: F, ) where - F: Fn(ConnectionId, Vec, bool) -> S, + F: Fn(SocketData) -> S, S: Debug, { log::trace!("Sending keepalive for connection: {connection_id}"); - let ordered_msg = message_sender.wrap_message(Vec::new()).into_bytes(); + let message = SocketData::new(message_sender.sequence(), connection_id, false, Vec::new()); + mix_sender - .send(adapter_fn(connection_id, ordered_msg, false)) + .send(adapter_fn(message)) .await .expect("BatchRealMessageReceiver has stopped receiving!"); } @@ -63,7 +64,7 @@ async fn deal_with_data( adapter_fn: F, ) -> bool where - F: Fn(ConnectionId, Vec, bool) -> S, + F: Fn(SocketData) -> S, S: Debug, { let (read_data, is_finished) = match read_data { @@ -86,15 +87,15 @@ where 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() + let message = SocketData::new( + message_sender.sequence(), + connection_id, + is_finished, + read_data.into(), ); mix_sender - .send(adapter_fn(connection_id, ordered_msg, is_finished)) + .send(adapter_fn(message)) .await .expect("InputMessageReceiver has stopped receiving!"); @@ -172,7 +173,7 @@ pub(super) async fn run_inbound( mut shutdown_listener: TaskClient, ) -> OwnedReadHalf where - F: Fn(ConnectionId, Vec, bool) -> S + Send + 'static, + F: Fn(SocketData) -> S + Send + 'static, S: Debug, { // TODO: this multiplication by 4 is completely arbitrary here diff --git a/common/socks5/proxy-helpers/src/proxy_runner/mod.rs b/common/socks5/proxy-helpers/src/proxy_runner/mod.rs index 65e24938a8..630ee13123 100644 --- a/common/socks5/proxy-helpers/src/proxy_runner/mod.rs +++ b/common/socks5/proxy-helpers/src/proxy_runner/mod.rs @@ -2,7 +2,7 @@ // SPDX-License-Identifier: Apache-2.0 use crate::connection_controller::ConnectionReceiver; -use nym_socks5_requests::ConnectionId; +use nym_socks5_requests::{ConnectionId, SocketData}; use nym_task::connections::LaneQueueLengths; use nym_task::TaskClient; use std::fmt::Debug; @@ -92,7 +92,7 @@ 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()); diff --git a/common/socks5/requests/src/lib.rs b/common/socks5/requests/src/lib.rs index d6cff2a5e3..ea0af9d841 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,126 @@ pub mod version; pub type Socks5ProviderRequest = interface::Request; pub type Socks5ProviderResponse = interface::Response; +#[derive(Debug, Error)] +#[error( + "didn't receive enough data to recover socket data. got {received}, but expected {expected}" +)] +pub struct InsufficientSocketDataError { + received: usize, + expected: usize, +} + +#[derive(Debug, Clone, Copy)] +pub struct SocketDataHeader { + pub seq: u64, + pub connection_id: ConnectionId, + pub local_socket_closed: bool, +} + +impl SocketDataHeader { + const SERIALIZED_LEN: usize = mem::size_of::() + mem::size_of::() + 1; + + pub fn try_from_bytes(b: &[u8]) -> Result { + if b.is_empty() { + return Err(InsufficientSocketDataError { + received: 0, + expected: Self::SERIALIZED_LEN, + }); + } + + 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 seq = u64::from_be_bytes(b[0..8].try_into().unwrap()); + let local_socket_closed = b[8] != 0; + let connection_id = ConnectionId::from_be_bytes(b[9..17].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_bytes(self) -> Vec { + self.into_bytes_iter().collect() + } + + pub fn into_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()) + } +} + +#[derive(Debug, Clone)] +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, + } + } + + pub fn try_from_bytes(b: &[u8]) -> Result { + 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, + }); + } + + let header = SocketDataHeader::try_from_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_bytes(self) -> Vec { + self.header + .into_bytes_iter() + .chain(self.data.into_iter()) + .collect() + } + + pub fn into_bytes_iter(self) -> impl Iterator { + self.header.into_bytes_iter().chain(self.data.into_iter()) + } +} + #[derive(Debug, Error)] pub enum Socks5RequestError { #[error("failed to deserialize received request: {source}")] @@ -99,9 +220,9 @@ 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.data.len(), 111); + 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..ae710ee6ea 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_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_bytes_iter()) .collect(), Socks5RequestContent::Query(query) => { diff --git a/common/socks5/requests/src/response.rs b/common/socks5/requests/src/response.rs index d0a34e074b..bd1a65dac5 100644 --- a/common/socks5/requests/src/response.rs +++ b/common/socks5/requests/src/response.rs @@ -239,6 +239,7 @@ impl Socks5ResponseContent { /// can be serialized and sent back through the mixnet to the requesting /// application. #[derive(Clone, Debug, PartialEq, Eq)] +// #[deprecated] pub struct NetworkData { pub data: Vec, pub connection_id: ConnectionId, diff --git a/service-providers/network-requester/src/core.rs b/service-providers/network-requester/src/core.rs index d948c79c6d..0b691ef282 100644 --- a/service-providers/network-requester/src/core.rs +++ b/service-providers/network-requester/src/core.rs @@ -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) diff --git a/service-providers/network-requester/src/socks5/tcp.rs b/service-providers/network-requester/src/socks5/tcp.rs index 2a50bbf3a3..018d75f185 100644 --- a/service-providers/network-requester/src/socks5/tcp.rs +++ b/service-providers/network-requester/src/socks5/tcp.rs @@ -66,13 +66,13 @@ 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.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..274836a9c2 100644 --- a/service-providers/network-requester/src/statistics/collector.rs +++ b/service-providers/network-requester/src/statistics/collector.rs @@ -9,7 +9,9 @@ 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, @@ -191,8 +193,9 @@ impl StatisticsCollector for ServiceStatisticsCollector { 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 ordered_msg = message_sender.wrap_message(msg).into_bytes(); + let message = SocketData::new(message_sender.sequence(), conn_id, true, msg); + let send_req = Socks5RequestContent::new_send(message); let mixnet_message = MixnetMessage::new_network_data_request( self.stats_provider_addr, From 5f9e54c83c6d03ecbd72936649961e502722bc7b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Thu, 8 Jun 2023 12:27:17 +0100 Subject: [PATCH 02/12] buggy, but backwards compatible, explicit sequencing --- .../src/socks/mixnet_responses.rs | 7 +- common/socks5/ordered-buffer/src/buffer.rs | 82 +++++---- common/socks5/ordered-buffer/src/lib.rs | 14 ++ .../src/connection_controller.rs | 100 +++++------ common/socks5/requests/src/lib.rs | 163 ++++++++++++++---- common/socks5/requests/src/request.rs | 4 +- common/socks5/requests/src/response.rs | 83 +++------ .../network-requester/src/core.rs | 2 +- .../network-requester/src/reply.rs | 3 +- .../network-requester/src/socks5/tcp.rs | 1 + 10 files changed, 270 insertions(+), 189 deletions(-) diff --git a/common/socks5-client-core/src/socks/mixnet_responses.rs b/common/socks5-client-core/src/socks/mixnet_responses.rs index d8ef425ddb..e554759fb0 100644 --- a/common/socks5-client-core/src/socks/mixnet_responses.rs +++ b/common/socks5-client-core/src/socks/mixnet_responses.rs @@ -1,7 +1,10 @@ +// 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, @@ -12,8 +15,6 @@ use nym_socks5_requests::{Socks5ProviderResponse, Socks5Response, Socks5Response use nym_sphinx::receiver::ReconstructedMessage; use nym_task::TaskClient; -use crate::error::Socks5ClientCoreError; - pub(crate) struct MixnetResponseListener { buffer_requester: ReceivedBufferRequestSender, mix_response_receiver: ReconstructedMessagesReceiver, diff --git a/common/socks5/ordered-buffer/src/buffer.rs b/common/socks5/ordered-buffer/src/buffer.rs index 33c4d69142..ec9272905f 100644 --- a/common/socks5/ordered-buffer/src/buffer.rs +++ b/common/socks5/ordered-buffer/src/buffer.rs @@ -1,6 +1,8 @@ -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; /// Stores messages and emits them in order. /// @@ -9,36 +11,49 @@ 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) { + // reject messages that have clearly malformed sequence + if sequence > self.next_sequence + MAX_REASONABLE_OFFSET { + error!("attempted to write message at index {sequence} while the next expected message was at only {}", self.next_sequence); + return; + } + + // TODO: make it return an error + if sequence < self.next_sequence { + panic!("received already reassembled message") + } + 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); } /// Returns `Option>` where it's `Some(bytes)` if there is gapless @@ -50,32 +65,29 @@ impl OrderedMessageBuffer { /// return `None` until message 3 comes in, at which point 3, 4, and any /// further contiguous messages which have arrived will be returned. 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, }) } } @@ -268,7 +280,7 @@ mod test_chunking_and_reassembling { buffer.write(two_message); buffer.write(one_message); assert!(buffer.read().is_some()); - assert_eq!(buffer.next_index, 3); + assert_eq!(buffer.next_sequence, 3); } #[test] @@ -296,23 +308,23 @@ mod test_chunking_and_reassembling { }; buffer.write(zero_message); assert!(buffer.read().is_some()); - assert_eq!(buffer.next_index, 1); + assert_eq!(buffer.next_sequence, 1); buffer.write(four_message); assert!(buffer.read().is_none()); - assert_eq!(buffer.next_index, 1); + assert_eq!(buffer.next_sequence, 1); buffer.write(three_message); assert!(buffer.read().is_none()); - assert_eq!(buffer.next_index, 1); + assert_eq!(buffer.next_sequence, 1); buffer.write(two_message); assert!(buffer.read().is_none()); - assert_eq!(buffer.next_index, 1); + assert_eq!(buffer.next_sequence, 1); buffer.write(one_message); 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..4e6de98d18 100644 --- a/common/socks5/ordered-buffer/src/lib.rs +++ b/common/socks5/ordered-buffer/src/lib.rs @@ -1,8 +1,22 @@ mod buffer; +#[deprecated] mod message; +#[deprecated] mod sender; pub use buffer::{OrderedMessageBuffer, ReadContiguousData}; pub use message::MessageError; pub use message::OrderedMessage; pub use sender::OrderedMessageSender; +use std::sync::atomic::AtomicU64; +use std::sync::Arc; + +#[derive(Debug, Clone)] +pub struct Socks5DataStream { + inner: Arc, +} + +#[derive(Debug)] +pub struct Socks5DataStreamInner { + next_seq: AtomicU64, +} diff --git a/common/socks5/proxy-helpers/src/connection_controller.rs b/common/socks5/proxy-helpers/src/connection_controller.rs index c1a9ff3f47..3e8a6d24f8 100644 --- a/common/socks5/proxy-helpers/src/connection_controller.rs +++ b/common/socks5/proxy-helpers/src/connection_controller.rs @@ -4,8 +4,8 @@ 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, NetworkData, SendRequest, SocketData}; use nym_task::connections::{ConnectionCommand, ConnectionCommandSender}; use nym_task::TaskClient; use std::collections::{HashMap, HashSet}; @@ -40,31 +40,25 @@ 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, + if value.inner.data.is_empty() { + println!("1: EMPTY SOCKET DATA") } + ControllerCommand::Send { data: value.inner } } } impl From for ControllerCommand { fn from(value: SendRequest) -> Self { - // TODO: do we care about sequence number here? - - ControllerCommand::Send { - connection_id: value.data.header.connection_id, - data: value.data.data, - is_closed: value.data.header.local_socket_closed, + if value.data.data.is_empty() { + println!("2: EMPTY SOCKET DATA") } + ControllerCommand::Send { data: value.data } } } @@ -76,18 +70,11 @@ 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); } - self.ordered_buffer.write(ordered_message); + self.ordered_buffer.write(seq, payload); } fn read_from_buf(&mut self) -> Option { @@ -119,7 +106,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, } @@ -156,8 +143,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) } } } @@ -186,17 +173,19 @@ 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 { + 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) { + if !message.data.is_empty() { + active_connection.write_to_buf(hdr.seq, message.data, hdr.local_socket_closed); + } else if !hdr.local_socket_closed { error!("Tried to write an empty message to a not-closing connection. Please let us know if you see this 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; } } @@ -220,23 +209,23 @@ impl Controller { // 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 ) } } @@ -244,20 +233,23 @@ impl Controller { pub async fn run(&mut self) { 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::Insert{connection_id, connection_sender}) => { - self.insert_connection(connection_id, connection_sender) - } - Some(ControllerCommand::Remove{ connection_id }) => self.remove_connection(connection_id), - None => { - log::trace!("SOCKS5 Controller: Stopping since channel closed"); - break; - } - }, + command = self.receiver.next() => match command { + Some(ControllerCommand::Send{data}) => { + if data.data.is_empty() { + println!("3: EMPTY SOCKET DATA") } + self.send_to_connection(data) + } + Some(ControllerCommand::Insert{connection_id, connection_sender}) => { + self.insert_connection(connection_id, connection_sender) + } + Some(ControllerCommand::Remove{ connection_id }) => self.remove_connection(connection_id), + None => { + log::trace!("SOCKS5 Controller: Stopping since channel closed"); + break; + } + }, + } } self.shutdown.recv_timeout().await; log::debug!("SOCKS5 Controller: Exiting"); diff --git a/common/socks5/requests/src/lib.rs b/common/socks5/requests/src/lib.rs index ea0af9d841..5d18d755d0 100644 --- a/common/socks5/requests/src/lib.rs +++ b/common/socks5/requests/src/lib.rs @@ -17,9 +17,9 @@ pub mod version; pub type Socks5ProviderRequest = interface::Request; pub type Socks5ProviderResponse = interface::Response; -#[derive(Debug, Error)] +#[derive(Debug, Error, PartialEq, Eq)] #[error( - "didn't receive enough data to recover socket data. got {received}, but expected {expected}" + "didn't receive enough data to recover socket data. got {received}, but expected at least {expected}" )] pub struct InsufficientSocketDataError { received: usize, @@ -34,16 +34,14 @@ pub struct SocketDataHeader { } impl SocketDataHeader { - const SERIALIZED_LEN: usize = mem::size_of::() + mem::size_of::() + 1; + const SERIALIZED_LEN: usize = mem::size_of::() + 1 + mem::size_of::(); - pub fn try_from_bytes(b: &[u8]) -> Result { - if b.is_empty() { - return Err(InsufficientSocketDataError { - received: 0, - expected: Self::SERIALIZED_LEN, - }); - } + // 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(), @@ -52,9 +50,76 @@ impl SocketDataHeader { } // the unwraps here are fine as we just ensured we have the exact amount of bytes we need - let seq = u64::from_be_bytes(b[0..8].try_into().unwrap()); + let connection_id = ConnectionId::from_be_bytes(b[0..8].try_into().unwrap()); let local_socket_closed = b[8] != 0; - let connection_id = ConnectionId::from_be_bytes(b[9..17].try_into().unwrap()); + let seq = u64::from_be_bytes(b[9..].try_into().unwrap()); + + Ok(SocketDataHeader { + seq, + connection_id, + local_socket_closed, + }) + + // + // let with_seq = mem::size_of::() + 1 + mem::size_of::(); + // let without_seq = mem::size_of::() + 1; + // + // if b.len() < without_seq { + // return Err(InsufficientSocketDataError { + // received: b.len(), + // expected: without_seq, + // }); + // } + // + // // 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; + // + // if b.len() > without_seq { + // return Err(InsufficientSocketDataError { + // received: b.len(), + // expected: without_seq, + // }); + // } + // + // let seq = u64::from_be_bytes(b[0..8].try_into().unwrap()); + // + // Ok(SocketDataHeader { + // seq, + // connection_id, + // local_socket_closed: 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, @@ -65,16 +130,14 @@ impl SocketDataHeader { // 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_bytes(self) -> Vec { - self.into_bytes_iter().collect() + // SOCKET_CLOSED (1B) || CONNECTION_ID (8B) || SEQUENCE (8B) + pub fn into_response_bytes(self) -> Vec { + self.into_response_bytes_iter().collect() } - pub fn into_bytes_iter(self) -> impl Iterator { - self.connection_id - .to_be_bytes() - .into_iter() - .chain(std::iter::once(self.local_socket_closed as u8)) + 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()) } } @@ -92,6 +155,10 @@ impl SocketData { local_socket_closed: bool, data: Vec, ) -> Self { + if data.is_empty() { + println!("making socket data with no data"); + println!("{local_socket_closed}"); + } SocketData { header: SocketDataHeader { seq, @@ -102,7 +169,7 @@ impl SocketData { } } - pub fn try_from_bytes(b: &[u8]) -> Result { + fn verify_deserialization_len(b: &[u8]) -> Result<(), InsufficientSocketDataError> { if b.is_empty() { return Err(InsufficientSocketDataError { received: 0, @@ -116,24 +183,60 @@ impl SocketData { expected: SocketDataHeader::SERIALIZED_LEN, }); } + Ok(()) + } - let header = SocketDataHeader::try_from_bytes(&b[..SocketDataHeader::SERIALIZED_LEN])?; - let data = b[..SocketDataHeader::SERIALIZED_LEN].to_vec(); + // 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])?; + println!( + "req: {} bytes of data ", + b[SocketDataHeader::SERIALIZED_LEN..].len() + ); + println!("seq: {}", header.seq); + 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_bytes(self) -> Vec { - self.header - .into_bytes_iter() - .chain(self.data.into_iter()) - .collect() + pub fn into_request_bytes(self) -> Vec { + self.into_request_bytes_iter().collect() } - pub fn into_bytes_iter(self) -> impl Iterator { - self.header.into_bytes_iter().chain(self.data.into_iter()) + 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])?; + println!( + "res: {} bytes of data ", + b[SocketDataHeader::SERIALIZED_LEN..].len() + ); + println!("seq: {}", header.seq); + 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()) } } diff --git a/common/socks5/requests/src/request.rs b/common/socks5/requests/src/request.rs index ae710ee6ea..50ed5b51a9 100644 --- a/common/socks5/requests/src/request.rs +++ b/common/socks5/requests/src/request.rs @@ -314,7 +314,7 @@ impl Socks5RequestContent { )) } RequestFlag::Send => Ok(Socks5RequestContent::Send(SendRequest { - data: SocketData::try_from_bytes(&b[1..])?, + data: SocketData::try_from_request_bytes(&b[1..])?, })), RequestFlag::Query => { use bincode::Options; @@ -346,7 +346,7 @@ impl Socks5RequestContent { } } Socks5RequestContent::Send(req) => std::iter::once(RequestFlag::Send as u8) - .chain(req.data.into_bytes_iter()) + .chain(req.data.into_request_bytes_iter()) .collect(), Socks5RequestContent::Query(query) => { diff --git a/common/socks5/requests/src/response.rs b/common/socks5/requests/src/response.rs index bd1a65dac5..941e19e573 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), } } @@ -166,15 +166,12 @@ pub enum Socks5ResponseContent { 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(NetworkData::new(seq, connection_id, data, is_closed)) } pub fn new_connection_error( @@ -239,67 +236,27 @@ impl Socks5ResponseContent { /// can be serialized and sent back through the mixnet to the requesting /// application. #[derive(Clone, Debug, PartialEq, Eq)] -// #[deprecated] pub struct NetworkData { - pub data: Vec, - pub connection_id: ConnectionId, - pub is_closed: bool, + pub inner: SocketData, } impl NetworkData { /// Constructor for responses - pub fn new(connection_id: ConnectionId, data: Vec, is_closed: bool) -> Self { + pub fn new(seq: u64, 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, + inner: SocketData::new(seq, connection_id, is_closed, data), } } 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) + let inner = SocketData::try_from_response_bytes(b)?; + Ok(NetworkData { inner }) } /// 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() + self.inner.into_response_bytes() } } diff --git a/service-providers/network-requester/src/core.rs b/service-providers/network-requester/src/core.rs index 0b691ef282..1503c088fb 100644 --- a/service-providers/network-requester/src/core.rs +++ b/service-providers/network-requester/src/core.rs @@ -359,7 +359,7 @@ impl NRServiceProvider { return_address, remote_version, connection_id, - NetworkData::new_closed_empty(connection_id), + NetworkData::new(0, connection_id, Vec::new(), true), ); mix_input_sender diff --git a/service-providers/network-requester/src/reply.rs b/service-providers/network-requester/src/reply.rs index 6fe313e30b..d1ec4a715a 100644 --- a/service-providers/network-requester/src/reply.rs +++ b/service-providers/network-requester/src/reply.rs @@ -135,11 +135,12 @@ 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); + let response_content = NetworkData::new(seq, connection_id, data, closed_socket); 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 018d75f185..baed05393d 100644 --- a/service-providers/network-requester/src/socks5/tcp.rs +++ b/service-providers/network-requester/src/socks5/tcp.rs @@ -70,6 +70,7 @@ impl Connection { MixnetMessage::new_network_data_response_content( return_address.clone(), remote_version.clone(), + socket_data.header.seq, socket_data.header.connection_id, socket_data.data, socket_data.header.local_socket_closed, From 45f3f3ec0156656ca73603830f140a9068520b90 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Thu, 8 Jun 2023 15:31:50 +0100 Subject: [PATCH 03/12] cleanup --- .../src/socks/mixnet_responses.rs | 6 +- .../src/connection_controller.rs | 76 +++++++------------ common/socks5/requests/src/lib.rs | 14 ---- common/socks5/requests/src/request.rs | 6 +- common/socks5/requests/src/response.rs | 44 +++-------- .../network-requester/src/core.rs | 8 +- .../network-requester/src/reply.rs | 11 ++- 7 files changed, 55 insertions(+), 110 deletions(-) diff --git a/common/socks5-client-core/src/socks/mixnet_responses.rs b/common/socks5-client-core/src/socks/mixnet_responses.rs index e554759fb0..05cb5bd48d 100644 --- a/common/socks5-client-core/src/socks/mixnet_responses.rs +++ b/common/socks5-client-core/src/socks/mixnet_responses.rs @@ -10,7 +10,7 @@ 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; @@ -80,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/proxy-helpers/src/connection_controller.rs b/common/socks5/proxy-helpers/src/connection_controller.rs index 3e8a6d24f8..b9f8c6dfd5 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::{OrderedMessageBuffer, ReadContiguousData}; -use nym_socks5_requests::{ConnectionId, NetworkData, SendRequest, SocketData}; +use nym_socks5_requests::{ConnectionId, SocketData}; use nym_task::connections::{ConnectionCommand, ConnectionCommandSender}; use nym_task::TaskClient; use std::collections::{HashMap, HashSet}; @@ -44,21 +44,9 @@ pub enum ControllerCommand { }, } -impl From for ControllerCommand { - fn from(value: NetworkData) -> Self { - if value.inner.data.is_empty() { - println!("1: EMPTY SOCKET DATA") - } - ControllerCommand::Send { data: value.inner } - } -} - -impl From for ControllerCommand { - fn from(value: SendRequest) -> Self { - if value.data.data.is_empty() { - println!("2: EMPTY SOCKET DATA") - } - ControllerCommand::Send { data: value.data } +impl ControllerCommand { + pub fn new_send(data: SocketData) -> Self { + ControllerCommand::Send { data } } } @@ -176,12 +164,8 @@ impl Controller { 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) { - if !message.data.is_empty() { - active_connection.write_to_buf(hdr.seq, message.data, hdr.local_socket_closed); - } else if !hdr.local_socket_closed { - error!("Tried to write an empty message to a not-closing connection. Please let us know if you see this message"); - active_connection.write_to_buf(hdr.seq, message.data, hdr.local_socket_closed); - } + // 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 { @@ -189,6 +173,13 @@ impl Controller { 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() @@ -198,16 +189,8 @@ 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(&hdr.connection_id) { debug!("Received a 'Send' before 'Connect' - going to buffer the data"); @@ -233,23 +216,20 @@ impl Controller { pub async fn run(&mut self) { loop { tokio::select! { - command = self.receiver.next() => match command { - Some(ControllerCommand::Send{data}) => { - if data.data.is_empty() { - println!("3: EMPTY SOCKET DATA") + command = self.receiver.next() => match command { + Some(ControllerCommand::Send{data}) => { + self.send_to_connection(data) + } + Some(ControllerCommand::Insert{connection_id, connection_sender}) => { + self.insert_connection(connection_id, connection_sender) + } + Some(ControllerCommand::Remove{ connection_id }) => self.remove_connection(connection_id), + None => { + log::trace!("SOCKS5 Controller: Stopping since channel closed"); + break; + } + }, } - self.send_to_connection(data) - } - Some(ControllerCommand::Insert{connection_id, connection_sender}) => { - self.insert_connection(connection_id, connection_sender) - } - Some(ControllerCommand::Remove{ connection_id }) => self.remove_connection(connection_id), - None => { - log::trace!("SOCKS5 Controller: Stopping since channel closed"); - break; - } - }, - } } self.shutdown.recv_timeout().await; log::debug!("SOCKS5 Controller: Exiting"); diff --git a/common/socks5/requests/src/lib.rs b/common/socks5/requests/src/lib.rs index 5d18d755d0..6b3ea48c91 100644 --- a/common/socks5/requests/src/lib.rs +++ b/common/socks5/requests/src/lib.rs @@ -155,10 +155,6 @@ impl SocketData { local_socket_closed: bool, data: Vec, ) -> Self { - if data.is_empty() { - println!("making socket data with no data"); - println!("{local_socket_closed}"); - } SocketData { header: SocketDataHeader { seq, @@ -192,11 +188,6 @@ impl SocketData { Self::verify_deserialization_len(b)?; let header = SocketDataHeader::try_from_request_bytes(&b[..SocketDataHeader::SERIALIZED_LEN])?; - println!( - "req: {} bytes of data ", - b[SocketDataHeader::SERIALIZED_LEN..].len() - ); - println!("seq: {}", header.seq); let data = b[SocketDataHeader::SERIALIZED_LEN..].to_vec(); Ok(SocketData { header, data }) @@ -219,11 +210,6 @@ impl SocketData { let header = SocketDataHeader::try_from_response_bytes(&b[..SocketDataHeader::SERIALIZED_LEN])?; - println!( - "res: {} bytes of data ", - b[SocketDataHeader::SERIALIZED_LEN..].len() - ); - println!("seq: {}", header.seq); let data = b[SocketDataHeader::SERIALIZED_LEN..].to_vec(); Ok(SocketData { header, data }) diff --git a/common/socks5/requests/src/request.rs b/common/socks5/requests/src/request.rs index 50ed5b51a9..4dfc096b30 100644 --- a/common/socks5/requests/src/request.rs +++ b/common/socks5/requests/src/request.rs @@ -231,7 +231,7 @@ impl Socks5RequestContent { /// Construct a new Request::Send instance pub fn new_send(data: SocketData) -> Socks5RequestContent { - Socks5RequestContent::Send(SendRequest { data }) + Socks5RequestContent::Send(SendRequest { data: data }) } /// Deserialize the request type, connection id, destination address and port, @@ -609,7 +609,7 @@ mod request_deserialization_tests { match request { Socks5RequestContent::Send(SendRequest { conn_id, - data, + data: data, local_closed, }) => { assert_eq!(u64::from_be_bytes([1, 2, 3, 4, 5, 6, 7, 8]), conn_id); @@ -644,7 +644,7 @@ mod request_deserialization_tests { match request { Socks5RequestContent::Send(SendRequest { conn_id, - data, + data: data, local_closed, }) => { assert_eq!(u64::from_be_bytes([1, 2, 3, 4, 5, 6, 7, 8]), conn_id); diff --git a/common/socks5/requests/src/response.rs b/common/socks5/requests/src/response.rs index 941e19e573..5d0696d715 100644 --- a/common/socks5/requests/src/response.rs +++ b/common/socks5/requests/src/response.rs @@ -159,7 +159,7 @@ impl Socks5Response { #[derive(Clone, Debug, PartialEq, Eq)] pub enum Socks5ResponseContent { - NetworkData(NetworkData), + NetworkData { content: SocketData }, ConnectionError(ConnectionError), Query(QueryResponse), } @@ -171,7 +171,9 @@ impl Socks5ResponseContent { data: Vec, is_closed: bool, ) -> Socks5ResponseContent { - Socks5ResponseContent::NetworkData(NetworkData::new(seq, connection_id, data, is_closed)) + Socks5ResponseContent::NetworkData { + content: SocketData::new(seq, connection_id, is_closed, data), + } } pub fn new_connection_error( @@ -183,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) => { @@ -217,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..])?, )), @@ -232,34 +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 inner: SocketData, -} - -impl NetworkData { - /// Constructor for responses - pub fn new(seq: u64, connection_id: ConnectionId, data: Vec, is_closed: bool) -> Self { - NetworkData { - inner: SocketData::new(seq, connection_id, is_closed, data), - } - } - - pub fn try_from_bytes(b: &[u8]) -> Result { - let inner = SocketData::try_from_response_bytes(b)?; - Ok(NetworkData { inner }) - } - - /// 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 { - self.inner.into_response_bytes() - } -} - #[derive(Clone, Debug, PartialEq, Eq)] pub struct ConnectionError { pub connection_id: ConnectionId, diff --git a/service-providers/network-requester/src/core.rs b/service-providers/network-requester/src/core.rs index 1503c088fb..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, }; @@ -359,7 +359,7 @@ impl NRServiceProvider { return_address, remote_version, connection_id, - NetworkData::new(0, connection_id, Vec::new(), true), + 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 d1ec4a715a..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); @@ -140,7 +140,10 @@ impl MixnetMessage { data: Vec, closed_socket: bool, ) -> Self { - let response_content = NetworkData::new(seq, 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) } From 7228331db67cd727116f7dabeea0a6b86f9c3972 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Thu, 8 Jun 2023 15:46:31 +0100 Subject: [PATCH 04/12] removed OrderedMessage --- common/socks5/ordered-buffer/src/lib.rs | 4 - common/socks5/ordered-buffer/src/message.rs | 143 -------------------- 2 files changed, 147 deletions(-) delete mode 100644 common/socks5/ordered-buffer/src/message.rs diff --git a/common/socks5/ordered-buffer/src/lib.rs b/common/socks5/ordered-buffer/src/lib.rs index 4e6de98d18..df587e1070 100644 --- a/common/socks5/ordered-buffer/src/lib.rs +++ b/common/socks5/ordered-buffer/src/lib.rs @@ -1,12 +1,8 @@ mod buffer; #[deprecated] -mod message; -#[deprecated] mod sender; pub use buffer::{OrderedMessageBuffer, ReadContiguousData}; -pub use message::MessageError; -pub use message::OrderedMessage; pub use sender::OrderedMessageSender; use std::sync::atomic::AtomicU64; use std::sync::Arc; 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); -} From f4cd37280860d96f70c976bb020b871ecb99ee29 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Thu, 8 Jun 2023 16:04:56 +0100 Subject: [PATCH 05/12] simplified keeping track of message sequence --- Cargo.lock | 1 - common/socks5/ordered-buffer/src/lib.rs | 15 --- common/socks5/proxy-helpers/src/lib.rs | 2 + .../proxy-helpers/src/ordered_sender.rs | 99 +++++++++++++++++ .../proxy-helpers/src/proxy_runner/inbound.rs | 104 ++---------------- .../network-requester/Cargo.toml | 1 - .../src/statistics/collector.rs | 6 +- 7 files changed, 110 insertions(+), 118 deletions(-) create mode 100644 common/socks5/proxy-helpers/src/ordered_sender.rs 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/ordered-buffer/src/lib.rs b/common/socks5/ordered-buffer/src/lib.rs index df587e1070..66c8079d45 100644 --- a/common/socks5/ordered-buffer/src/lib.rs +++ b/common/socks5/ordered-buffer/src/lib.rs @@ -1,18 +1,3 @@ mod buffer; -#[deprecated] -mod sender; pub use buffer::{OrderedMessageBuffer, ReadContiguousData}; -pub use sender::OrderedMessageSender; -use std::sync::atomic::AtomicU64; -use std::sync::Arc; - -#[derive(Debug, Clone)] -pub struct Socks5DataStream { - inner: Arc, -} - -#[derive(Debug)] -pub struct Socks5DataStreamInner { - next_seq: AtomicU64, -} diff --git a/common/socks5/proxy-helpers/src/lib.rs b/common/socks5/proxy-helpers/src/lib.rs index cce784c4da..4df7af7751 100644 --- a/common/socks5/proxy-helpers/src/lib.rs +++ b/common/socks5/proxy-helpers/src/lib.rs @@ -4,3 +4,5 @@ pub mod available_reader; pub mod connection_controller; pub mod proxy_runner; +pub mod ordered_sender; + 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..d3d1e7549a --- /dev/null +++ b/common/socks5/proxy-helpers/src/ordered_sender.rs @@ -0,0 +1,99 @@ +// 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 struct OrderedMessageSender { + connection_id: ConnectionId, + mixnet_sender: MixProxySender, + + next_message_seq: u64, + mix_message_adapter: F, +} + +impl OrderedMessageSender +where + F: Fn(SocketData) -> S, +{ + pub fn new( + connection_id: ConnectionId, + mixnet_sender: MixProxySender, + mix_message_adapter: F, + ) -> Self { + OrderedMessageSender { + 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 async fn send_empty_close(&mut self) { + let message = self.construct_message(Vec::new(), true); + self.send_message(message).await + } + + pub 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 async fn deal_with_data( + &mut self, + read_data: Option>, + local_destination_address: &str, + remote_source_address: &str, + ) -> bool { + let connection_id = self.connection_id; + 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 + ); + let message = self.construct_message(read_data.into(), is_finished); + self.send_message(message).await; + + is_finished + } +} diff --git a/common/socks5/proxy-helpers/src/proxy_runner/inbound.rs b/common/socks5/proxy-helpers/src/proxy_runner/inbound.rs index e6947d4c25..6552714031 100644 --- a/common/socks5/proxy-helpers/src/proxy_runner/inbound.rs +++ b/common/socks5/proxy-helpers/src/proxy_runner/inbound.rs @@ -4,104 +4,21 @@ 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, 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(SocketData) -> S, - S: Debug, -{ - let message = SocketData::new(message_sender.sequence(), connection_id, true, Vec::new()); - mix_sender - .send(adapter_fn(message)) - .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(SocketData) -> S, - S: Debug, -{ - log::trace!("Sending keepalive for connection: {connection_id}"); - let message = SocketData::new(message_sender.sequence(), connection_id, false, Vec::new()); - - mix_sender - .send(adapter_fn(message)) - .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(SocketData) -> 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 - ); - - let message = SocketData::new( - message_sender.sequence(), - connection_id, - is_finished, - read_data.into(), - ); - - mix_sender - .send(adapter_fn(message)) - .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( @@ -179,7 +96,7 @@ where // 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(); + let mut message_sender = OrderedMessageSender::new(connection_id, mix_sender, adapter_fn); // Shutdown if outbound signled to shutdown let shutdown_future = shutdown_notify.notified().then(|_| sleep(SHUTDOWN_TIMEOUT)); @@ -218,7 +135,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() => { @@ -234,7 +151,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 @@ -242,15 +159,8 @@ 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 { + if message_sender.deal_with_data(read_data, &local_destination_address, + &remote_source_address).await { // 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/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/statistics/collector.rs b/service-providers/network-requester/src/statistics/collector.rs index 274836a9c2..e0169165fb 100644 --- a/service-providers/network-requester/src/statistics/collector.rs +++ b/service-providers/network-requester/src/statistics/collector.rs @@ -6,7 +6,6 @@ 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::{ @@ -192,9 +191,8 @@ 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 message = SocketData::new(message_sender.sequence(), conn_id, true, msg); + + let message = SocketData::new(0, conn_id, true, msg); let send_req = Socks5RequestContent::new_send(message); let mixnet_message = MixnetMessage::new_network_data_request( From 8b046d4139b60948abae119a8a90bed6200d9adf Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Thu, 8 Jun 2023 16:21:47 +0100 Subject: [PATCH 06/12] split up functions a bit more --- .../proxy-helpers/src/ordered_sender.rs | 65 ++++++++++++------- .../proxy-helpers/src/proxy_runner/inbound.rs | 18 ++--- .../proxy-helpers/src/proxy_runner/mod.rs | 11 +++- 3 files changed, 56 insertions(+), 38 deletions(-) diff --git a/common/socks5/proxy-helpers/src/ordered_sender.rs b/common/socks5/proxy-helpers/src/ordered_sender.rs index d3d1e7549a..d71f843558 100644 --- a/common/socks5/proxy-helpers/src/ordered_sender.rs +++ b/common/socks5/proxy-helpers/src/ordered_sender.rs @@ -7,24 +7,31 @@ use log::{debug, error}; use nym_socks5_requests::{ConnectionId, SocketData}; use std::io; -pub struct OrderedMessageSender { +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 +impl OrderedMessageSender where F: Fn(SocketData) -> S, { - pub fn new( + 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, @@ -54,46 +61,56 @@ where } } - pub async fn send_empty_close(&mut self) { + pub(crate) async fn send_empty_close(&mut self) { let message = self.construct_message(Vec::new(), true); self.send_message(message).await } - pub async fn send_empty_keepalive(&mut self) { + 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 async fn deal_with_data( - &mut self, - read_data: Option>, - local_destination_address: &str, - remote_source_address: &str, - ) -> bool { - let connection_id = self.connection_id; + 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!("({connection_id}) socks5 inbound"), "failed to read request from the socket - {err}"); + error!(target: &*format!("({}) socks5 inbound", self.connection_id), "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 - ); - let message = self.construct_message(read_data.into(), is_finished); - self.send_message(message).await; + ProcessedData { + data: read_data, + is_done: is_finished, + } + } - 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 6552714031..82e6396a5f 100644 --- a/common/socks5/proxy-helpers/src/proxy_runner/inbound.rs +++ b/common/socks5/proxy-helpers/src/proxy_runner/inbound.rs @@ -1,7 +1,6 @@ // 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; @@ -13,7 +12,6 @@ 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 tokio::select; @@ -76,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(SocketData) -> S + Send + 'static, - S: Debug, { // 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(connection_id, mix_sender, adapter_fn); // Shutdown if outbound signled to shutdown let shutdown_future = shutdown_notify.notified().then(|_| sleep(SHUTDOWN_TIMEOUT)); @@ -159,8 +151,12 @@ where read_data = wait_until_lane_almost_empty(&lane_queue_lengths, connection_id) .then(|_| available_reader.next()), if !we_are_closed => { - if message_sender.deal_with_data(read_data, &local_destination_address, - &remote_source_address).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 630ee13123..38499ea59c 100644 --- a/common/socks5/proxy-helpers/src/proxy_runner/mod.rs +++ b/common/socks5/proxy-helpers/src/proxy_runner/mod.rs @@ -2,6 +2,7 @@ // SPDX-License-Identifier: Apache-2.0 use crate::connection_controller::ConnectionReceiver; +use crate::ordered_sender::OrderedMessageSender; use nym_socks5_requests::{ConnectionId, SocketData}; use nym_task::connections::LaneQueueLengths; use nym_task::TaskClient; @@ -98,14 +99,18 @@ where 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(), From 1328ba35be6fa9dc0ac2cedc5e5407b455640a89 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Thu, 8 Jun 2023 16:22:16 +0100 Subject: [PATCH 07/12] removed original Ordered sender --- common/socks5/ordered-buffer/src/sender.rs | 59 ---------------------- 1 file changed, 59 deletions(-) delete mode 100644 common/socks5/ordered-buffer/src/sender.rs diff --git a/common/socks5/ordered-buffer/src/sender.rs b/common/socks5/ordered-buffer/src/sender.rs deleted file mode 100644 index 40da98ef26..0000000000 --- a/common/socks5/ordered-buffer/src/sender.rs +++ /dev/null @@ -1,59 +0,0 @@ -/// 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_sequence: u64, -} - -impl OrderedMessageSender { - pub fn new() -> OrderedMessageSender { - OrderedMessageSender { next_sequence: 0 } - } - - pub fn sequence(&mut self) -> u64 { - let next = self.next_sequence; - self.next_sequence += 1; - next - } - - // /// 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_sequence, - // }; - // self.next_sequence += 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_message = sender.sequence(); - - assert_eq!(first_message, 0); - - let second_message = sender.sequence(); - assert_eq!(second_message, 1); - } - } -} From a8bf690c176f63c7ee38ffb4c5245a47c9bd1975 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Thu, 8 Jun 2023 16:52:33 +0100 Subject: [PATCH 08/12] tests --- common/socks5/ordered-buffer/src/buffer.rs | 242 ++++++++++-------- common/socks5/ordered-buffer/src/lib.rs | 2 +- .../src/connection_controller.rs | 4 +- 3 files changed, 134 insertions(+), 114 deletions(-) diff --git a/common/socks5/ordered-buffer/src/buffer.rs b/common/socks5/ordered-buffer/src/buffer.rs index ec9272905f..1b7b281727 100644 --- a/common/socks5/ordered-buffer/src/buffer.rs +++ b/common/socks5/ordered-buffer/src/buffer.rs @@ -3,6 +3,19 @@ use log::*; 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. /// @@ -35,16 +48,24 @@ impl OrderedMessageBuffer { /// 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, sequence: u64, data: Vec) { + 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 { - error!("attempted to write message at index {sequence} while the next expected message was at only {}", self.next_sequence); - return; + return Err(OrderedMessageError::MessageSequenceTooLarge { + current: self.next_sequence, + received: sequence, + }); + } + + if self.messages.contains_key(&sequence) { + return Err(OrderedMessageError::AttemptedToOverwriteSequence { received: sequence }); } - // TODO: make it return an error if sequence < self.next_sequence { - panic!("received already reassembled message") + return Err(OrderedMessageError::MessageAlreadyReconstructed { + current: self.next_sequence, + received: sequence, + }); } trace!( @@ -54,6 +75,7 @@ impl OrderedMessageBuffer { ); self.messages.insert(sequence, data); + Ok(()) } /// Returns `Option>` where it's `Some(bytes)` if there is gapless @@ -64,6 +86,7 @@ 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_sequence) { return None; @@ -102,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::*; @@ -114,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); @@ -138,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 @@ -158,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 @@ -182,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] @@ -211,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(), @@ -261,24 +302,15 @@ 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_sequence, 3); } @@ -286,43 +318,29 @@ mod test_chunking_and_reassembling { #[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_sequence, 1); - buffer.write(four_message); + buffer.write(4, four_message).unwrap(); assert!(buffer.read().is_none()); 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_sequence, 1); - buffer.write(two_message); + buffer.write(2, two_message).unwrap(); assert!(buffer.read().is_none()); 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_sequence, 5) } diff --git a/common/socks5/ordered-buffer/src/lib.rs b/common/socks5/ordered-buffer/src/lib.rs index 66c8079d45..97e5a76c1f 100644 --- a/common/socks5/ordered-buffer/src/lib.rs +++ b/common/socks5/ordered-buffer/src/lib.rs @@ -1,3 +1,3 @@ mod buffer; -pub use buffer::{OrderedMessageBuffer, ReadContiguousData}; +pub use buffer::{OrderedMessageBuffer, OrderedMessageError, ReadContiguousData}; diff --git a/common/socks5/proxy-helpers/src/connection_controller.rs b/common/socks5/proxy-helpers/src/connection_controller.rs index b9f8c6dfd5..0d93009626 100644 --- a/common/socks5/proxy-helpers/src/connection_controller.rs +++ b/common/socks5/proxy-helpers/src/connection_controller.rs @@ -62,7 +62,9 @@ impl ActiveConnection { if is_closed { self.closed_at_index = Some(seq); } - self.ordered_buffer.write(seq, payload); + if let Err(err) = self.ordered_buffer.write(seq, payload) { + error!("failed to write to the buffer: {err}") + } } fn read_from_buf(&mut self) -> Option { From f2c5dbb6966322ec24be6d5e655261646dee794d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Thu, 8 Jun 2023 16:53:55 +0100 Subject: [PATCH 09/12] deadcode --- common/socks5/requests/src/lib.rs | 30 ------------------------------ 1 file changed, 30 deletions(-) diff --git a/common/socks5/requests/src/lib.rs b/common/socks5/requests/src/lib.rs index 6b3ea48c91..661f7266e2 100644 --- a/common/socks5/requests/src/lib.rs +++ b/common/socks5/requests/src/lib.rs @@ -59,36 +59,6 @@ impl SocketDataHeader { connection_id, local_socket_closed, }) - - // - // let with_seq = mem::size_of::() + 1 + mem::size_of::(); - // let without_seq = mem::size_of::() + 1; - // - // if b.len() < without_seq { - // return Err(InsufficientSocketDataError { - // received: b.len(), - // expected: without_seq, - // }); - // } - // - // // 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; - // - // if b.len() > without_seq { - // return Err(InsufficientSocketDataError { - // received: b.len(), - // expected: without_seq, - // }); - // } - // - // let seq = u64::from_be_bytes(b[0..8].try_into().unwrap()); - // - // Ok(SocketDataHeader { - // seq, - // connection_id, - // local_socket_closed: local_socket_closed, - // }) } // the serialization of the header looks as follows: From 687b437ea0dc5649eac44858beeaf1a3215211b1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Thu, 8 Jun 2023 16:55:23 +0100 Subject: [PATCH 10/12] cargo fmt --- common/socks5/proxy-helpers/src/lib.rs | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/common/socks5/proxy-helpers/src/lib.rs b/common/socks5/proxy-helpers/src/lib.rs index 4df7af7751..673dd9ea46 100644 --- a/common/socks5/proxy-helpers/src/lib.rs +++ b/common/socks5/proxy-helpers/src/lib.rs @@ -3,6 +3,5 @@ pub mod available_reader; pub mod connection_controller; -pub mod proxy_runner; pub mod ordered_sender; - +pub mod proxy_runner; From 182e147a8654c2ff279de2a919c4607f9338232a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Thu, 8 Jun 2023 17:31:17 +0100 Subject: [PATCH 11/12] fixed and updated related units tests --- common/socks5/requests/src/lib.rs | 104 ++++++++++++++++++++++++- common/socks5/requests/src/request.rs | 84 +++++++++++++------- common/socks5/requests/src/response.rs | 51 ------------ 3 files changed, 157 insertions(+), 82 deletions(-) diff --git a/common/socks5/requests/src/lib.rs b/common/socks5/requests/src/lib.rs index 661f7266e2..58c23dcd9d 100644 --- a/common/socks5/requests/src/lib.rs +++ b/common/socks5/requests/src/lib.rs @@ -26,7 +26,7 @@ pub struct InsufficientSocketDataError { expected: usize, } -#[derive(Debug, Clone, Copy)] +#[derive(Debug, Clone, Copy, Eq, PartialEq)] pub struct SocketDataHeader { pub seq: u64, pub connection_id: ConnectionId, @@ -112,7 +112,7 @@ impl SocketDataHeader { } } -#[derive(Debug, Clone)] +#[derive(Debug, Clone, PartialEq, Eq)] pub struct SocketData { pub header: SocketDataHeader, pub data: Vec, @@ -231,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::*; @@ -280,7 +377,8 @@ mod tests { RequestContent::ProviderData(req) => match req.content { Socks5RequestContent::Send(send_req) => { assert_eq!(send_req.data.header.connection_id, 7810961472501196273); - assert_eq!(send_req.data.data.len(), 111); + 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 4dfc096b30..248d76442c 100644 --- a/common/socks5/requests/src/request.rs +++ b/common/socks5/requests/src/request.rs @@ -603,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: 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, @@ -634,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, @@ -642,14 +670,14 @@ mod request_deserialization_tests { let request = Socks5RequestContent::try_from_bytes(&request_bytes).unwrap(); match request { - Socks5RequestContent::Send(SendRequest { - conn_id, - data: 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 5d0696d715..4e0c906439 100644 --- a/common/socks5/requests/src/response.rs +++ b/common/socks5/requests/src/response.rs @@ -298,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::*; From bdcdcf7f8b7ff5cf957d8c45f02763d2a2d98873 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Fri, 9 Jun 2023 09:41:05 +0100 Subject: [PATCH 12/12] clippy --- common/socks5/requests/src/request.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/common/socks5/requests/src/request.rs b/common/socks5/requests/src/request.rs index 248d76442c..0e29fb49d9 100644 --- a/common/socks5/requests/src/request.rs +++ b/common/socks5/requests/src/request.rs @@ -231,7 +231,7 @@ impl Socks5RequestContent { /// Construct a new Request::Send instance pub fn new_send(data: SocketData) -> Socks5RequestContent { - Socks5RequestContent::Send(SendRequest { data: data }) + Socks5RequestContent::Send(SendRequest { data }) } /// Deserialize the request type, connection id, destination address and port,