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,