GRIN/nym
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

initial work on putting data sequence explicitly inside socks5 request

Browse Source
This commit is contained in:
Jędrzej Stuczyński
2023-06-07 14:03:44 +01:00
parent 71fb6a1ba1
commit f8c2f90502
11 changed files with 210 additions and 99 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/socks5-client-core/src/socks/client.rs
+4 -8
View File
@@ -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,
common/socks5/ordered-buffer/src/sender.rs
+21 -19
View File
@@ -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<u8>) -> 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<u8>) -> 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);
}
}
}
common/socks5/proxy-helpers/src/connection_controller.rs
+5 -3
View File
@@ -58,10 +58,12 @@ impl From<NetworkData> for ControllerCommand {
impl From<SendRequest> 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,
}
}
}
common/socks5/proxy-helpers/src/proxy_runner/inbound.rs
+17 -16
View File
@@ -1,4 +1,4 @@
// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
// Copyright 2021-2023 - Nym Technologies SA <contact@nymtech.net>
// 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<F, S>(
mix_sender: &MixProxySender<S>,
adapter_fn: F,
) where
F: Fn(ConnectionId, Vec<u8>, 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<F, S>(
mix_sender: &MixProxySender<S>,
adapter_fn: F,
) where
F: Fn(ConnectionId, Vec<u8>, 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<F, S>(
adapter_fn: F,
) -> bool
where
F: Fn(ConnectionId, Vec<u8>, 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<F, S>(
mut shutdown_listener: TaskClient,
) -> OwnedReadHalf
where
F: Fn(ConnectionId, Vec<u8>, bool) -> S + Send + 'static,
F: Fn(SocketData) -> S + Send + 'static,
S: Debug,
{
// TODO: this multiplication by 4 is completely arbitrary here
common/socks5/proxy-helpers/src/proxy_runner/mod.rs
+2 -2
View File
@@ -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<F>(mut self, adapter_fn: F) -> Self
where
F: Fn(ConnectionId, Vec<u8>, 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());
common/socks5/requests/src/lib.rs
+124 -3
View File
@@ -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<Socks5Request>;
pub type Socks5ProviderResponse = interface::Response<Socks5Request>;
#[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::<u64>() + mem::size_of::<ConnectionId>() + 1;
pub fn try_from_bytes(b: &[u8]) -> Result<SocketDataHeader, InsufficientSocketDataError> {
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<u8> {
self.into_bytes_iter().collect()
}
pub fn into_bytes_iter(self) -> impl Iterator<Item = u8> {
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<u8>,
}
impl SocketData {
pub fn new(
seq: u64,
connection_id: ConnectionId,
local_socket_closed: bool,
data: Vec<u8>,
) -> Self {
SocketData {
header: SocketDataHeader {
seq,
connection_id,
local_socket_closed,
},
data,
}
}
pub fn try_from_bytes(b: &[u8]) -> Result<SocketData, InsufficientSocketDataError> {
if b.is_empty() {
return Err(InsufficientSocketDataError {
received: 0,
expected: SocketDataHeader::SERIALIZED_LEN,
});
}
if b.len() < SocketDataHeader::SERIALIZED_LEN {
return Err(InsufficientSocketDataError {
received: b.len(),
expected: SocketDataHeader::SERIALIZED_LEN,
});
}
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<u8> {
self.header
.into_bytes_iter()
.chain(self.data.into_iter())
.collect()
}
pub fn into_bytes_iter(self) -> impl Iterator<Item = u8> {
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"),
},
common/socks5/requests/src/request.rs
+24 -39
View File
@@ -1,7 +1,10 @@
// Copyright 2020-2023 - Nym Technologies SA <contact@nymtech.net>
// 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<u8>,
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<u8>,
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<u8>,
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 || <RETURN_ADDR>
//
// send:
// RequestFlag::Send || CONN_ID || LOCAL_CLOSED || DATA
// where DATA: SEQ || TRUE_DATA
pub fn try_from_bytes(b: &[u8]) -> Result<Socks5RequestContent, RequestDeserializationError> {
// 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) => {
common/socks5/requests/src/response.rs
+1
View File
@@ -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<u8>,
pub connection_id: ConnectionId,
service-providers/network-requester/src/core.rs
+2 -2
View File
@@ -137,13 +137,13 @@ impl ServiceProvider<Socks5Request> 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)
service-providers/network-requester/src/socks5/tcp.rs
+4 -4
View File
@@ -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
service-providers/network-requester/src/statistics/collector.rs
+6 -3
View File
@@ -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,