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

cleanup

Browse Source
This commit is contained in:
Jędrzej Stuczyński
2023-06-08 15:31:50 +01:00
parent 5f9e54c83c
commit 45f3f3ec01
7 changed files with 55 additions and 110 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/socks5-client-core/src/socks/mixnet_responses.rs
+3 -3
View File
@@ -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(())
}
common/socks5/proxy-helpers/src/connection_controller.rs
+28 -48
View File
@@ -1,11 +1,11 @@
// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
// Copyright 2021-2023 - Nym Technologies SA <contact@nymtech.net>
// 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<NetworkData> 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<SendRequest> 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");
common/socks5/requests/src/lib.rs
-14
View File
@@ -155,10 +155,6 @@ impl SocketData {
local_socket_closed: bool,
data: Vec<u8>,
) -> 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 })
common/socks5/requests/src/request.rs
+3 -3
View File
@@ -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);
common/socks5/requests/src/response.rs
+9 -35
View File
@@ -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<u8>,
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<u8> {
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<u8>, is_closed: bool) -> Self {
NetworkData {
inner: SocketData::new(seq, connection_id, is_closed, data),
}
}
pub fn try_from_bytes(b: &[u8]) -> Result<NetworkData, ResponseDeserializationError> {
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<u8> {
self.inner.into_response_bytes()
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ConnectionError {
pub connection_id: ConnectionId,
service-providers/network-requester/src/core.rs
+5 -3
View File
@@ -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(
service-providers/network-requester/src/reply.rs
+7 -4
View File
@@ -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<Socks5Request>,
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<u8>,
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)
}