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

Rebase on develop

Browse Source 
commit 342883fcbe
Author: durch <durch@users.noreply.github.com>
Date:   Thu Apr 27 09:17:18 2023 +0200

    Put back PacketType 1

commit 61a0ee5a19
Author: Tommy Verrall <tommyvez@protonmail.com>
Date:   Wed Apr 26 16:37:29 2023 +0100

    change output for cpu-cycle management logs

commit 3956109c7e
Author: Tommy Verrall <tommy@nymtech.net>
Date:   Wed Apr 26 12:13:22 2023 +0100

    change the workflow file to build with cpucycles

commit 8d725b13c5
Author: durch <durch@users.noreply.github.com>
Date:   Mon Apr 24 13:14:58 2023 +0200

    Outfox client compat

commit 4d166c389b
Author: durch <durch@users.noreply.github.com>
Date:   Fri Apr 21 00:30:46 2023 +0200

    Address PR comments

commit 145c3c1223
Author: durch <durch@users.noreply.github.com>
Date:   Fri Apr 21 00:12:35 2023 +0200

    Rename PacketMode

commit cbd654d6fd
Author: Drazen Urch <drazen@urch.eu>
Date:   Thu Apr 20 23:59:40 2023 +0200

    Outfox rest compat (#3333)

    * Outfox forwarding compat

    * Tidy up interface

    * PacketSize compat

commit e7be91a94c
Author: durch <durch@users.noreply.github.com>
Date:   Wed Apr 19 16:36:48 2023 +0200

    Remove serde cruft

commit 582e7d566a
Author: durch <durch@users.noreply.github.com>
Date:   Wed Apr 19 16:24:09 2023 +0200

    Outfox framing

commit 6464da5f01
Author: durch <durch@users.noreply.github.com>
Date:   Tue Apr 18 22:23:02 2023 +0200

    Framing compat

commit d5e77e499b
Author: durch <durch@users.noreply.github.com>
Date:   Tue Apr 18 18:18:54 2023 +0200

    Framed encoding serde POC

commit f086f9c35a
Author: durch <durch@users.noreply.github.com>
Date:   Tue Apr 18 16:54:21 2023 +0200

    Experiment with serde
This commit is contained in:
durch
2023-04-27 09:45:25 +02:00
parent b4c525b664
commit f0de459c58
35 changed files with 284 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/build-and-upload-binaries-ci.yml
+1 -1
View File
@@ -69,7 +69,7 @@ jobs:
uses: actions-rs/cargo@v1
with:
command: build
args: --workspace --release --all
args: --workspace --release --all --features cpucycles
- name: Install Rust stable
uses: actions-rs/toolchain@v1
Cargo.lock
Generated
+1
View File
@@ -3786,6 +3786,7 @@ dependencies = [
"log",
"nym-crypto",
"nym-sphinx",
"nym-sphinx-params",
"nym-task",
"nym-topology",
"rand 0.7.3",
clients/native/src/client/mod.rs
+12 -4
View File
@@ -16,6 +16,7 @@ use nym_client_core::client::received_buffer::{
};
use nym_client_core::config::persistence::key_pathfinder::ClientKeyPathfinder;
use nym_sphinx::anonymous_replies::requests::AnonymousSenderTag;
use nym_sphinx::params::PacketType;
use nym_task::connections::TransmissionLane;
use nym_task::TaskManager;
use nym_validator_client::nyxd::QueryNyxdClient;
@@ -119,6 +120,7 @@ impl SocketClient {
self_address,
shared_lane_queue_lengths,
reply_controller_sender,
None,
);
websocket::Listener::new(config.get_listening_ip(), config.get_listening_port())
@@ -178,7 +180,10 @@ impl SocketClient {
Ok(started_client.task_manager)
}
pub async fn start_direct(self) -> Result<DirectClient, ClientError> {
pub async fn start_direct(
self,
packet_type: Option<PacketType>,
) -> Result<DirectClient, ClientError> {
if self.config.get_socket_type().is_websocket() {
return Err(ClientError::InvalidSocketMode);
}
@@ -224,6 +229,7 @@ impl SocketClient {
reconstructed_receiver,
address,
shutdown_notifier: started_client.task_manager,
packet_type,
})
}
}
@@ -237,6 +243,7 @@ pub struct DirectClient {
// we need to keep reference to this guy otherwise things will start dropping
shutdown_notifier: TaskManager,
packet_type: Option<PacketType>,
}
impl DirectClient {
@@ -257,7 +264,7 @@ impl DirectClient {
/// well enough in local tests)
pub async fn send_regular_message(&mut self, recipient: Recipient, message: Vec<u8>) {
let lane = TransmissionLane::General;
let input_msg = InputMessage::new_regular(recipient, message, lane);
let input_msg = InputMessage::new_regular(recipient, message, lane, self.packet_type);
self.client_input
.input_sender
@@ -276,7 +283,8 @@ impl DirectClient {
reply_surbs: u32,
) {
let lane = TransmissionLane::General;
let input_msg = InputMessage::new_anonymous(recipient, message, reply_surbs, lane);
let input_msg =
InputMessage::new_anonymous(recipient, message, reply_surbs, lane, self.packet_type);
self.client_input
.input_sender
@@ -290,7 +298,7 @@ impl DirectClient {
/// well enough in local tests)
pub async fn send_reply(&mut self, recipient_tag: AnonymousSenderTag, message: Vec<u8>) {
let lane = TransmissionLane::General;
let input_msg = InputMessage::new_reply(recipient_tag, message, lane);
let input_msg = InputMessage::new_reply(recipient_tag, message, lane, self.packet_type);
self.client_input
.input_sender
clients/native/src/websocket/handler.rs
+10 -3
View File
@@ -14,6 +14,7 @@ use nym_client_core::client::{
use nym_client_websocket_requests::{requests::ClientRequest, responses::ServerResponse};
use nym_sphinx::addressing::clients::Recipient;
use nym_sphinx::anonymous_replies::requests::AnonymousSenderTag;
use nym_sphinx::params::PacketType;
use nym_sphinx::receiver::ReconstructedMessage;
use nym_task::connections::{
ConnectionCommand, ConnectionCommandSender, ConnectionId, LaneQueueLengths, TransmissionLane,
@@ -41,6 +42,7 @@ pub(crate) struct HandlerBuilder {
self_full_address: Recipient,
lane_queue_lengths: LaneQueueLengths,
reply_controller_sender: ReplyControllerSender,
packet_type: Option<PacketType>,
}
impl HandlerBuilder {
@@ -51,6 +53,7 @@ impl HandlerBuilder {
self_full_address: &Recipient,
lane_queue_lengths: LaneQueueLengths,
reply_controller_sender: ReplyControllerSender,
packet_type: Option<PacketType>,
) -> Self {
Self {
msg_input,
@@ -59,6 +62,7 @@ impl HandlerBuilder {
self_full_address: *self_full_address,
lane_queue_lengths,
reply_controller_sender,
packet_type,
}
}
@@ -73,6 +77,7 @@ impl HandlerBuilder {
received_response_type: Default::default(),
lane_queue_lengths: self.lane_queue_lengths.clone(),
reply_controller_sender: self.reply_controller_sender.clone(),
packet_type: self.packet_type,
}
}
}
@@ -86,6 +91,7 @@ pub(crate) struct Handler {
received_response_type: ReceivedResponseType,
lane_queue_lengths: LaneQueueLengths,
reply_controller_sender: ReplyControllerSender,
packet_type: Option<PacketType>,
}
impl Drop for Handler {
@@ -160,7 +166,7 @@ impl Handler {
});
// the ack control is now responsible for chunking, etc.
let input_msg = InputMessage::new_regular(recipient, message, lane);
let input_msg = InputMessage::new_regular(recipient, message, lane, self.packet_type);
self.msg_input
.send(input_msg)
.await
@@ -191,7 +197,8 @@ impl Handler {
TransmissionLane::ConnectionId(id)
});
let input_msg = InputMessage::new_anonymous(recipient, message, reply_surbs, lane);
let input_msg =
InputMessage::new_anonymous(recipient, message, reply_surbs, lane, self.packet_type);
self.msg_input
.send(input_msg)
.await
@@ -218,7 +225,7 @@ impl Handler {
TransmissionLane::ConnectionId(id)
});
let input_msg = InputMessage::new_reply(recipient_tag, message, lane);
let input_msg = InputMessage::new_reply(recipient_tag, message, lane, self.packet_type);
self.msg_input
.send(input_msg)
.await
clients/webassembly/src/client/mod.rs
+11 -4
View File
@@ -21,6 +21,7 @@ use nym_client_core::config::{
CoverTraffic, DebugConfig, GatewayEndpointConfig, Topology, Traffic,
};
use nym_credential_storage::ephemeral_storage::EphemeralStorage;
use nym_sphinx::params::PacketType;
use nym_task::connections::TransmissionLane;
use nym_task::TaskManager;
use nym_topology::provider_trait::{HardcodedTopologyProvider, TopologyProvider};
@@ -49,6 +50,7 @@ pub struct NymClient {
// even though we don't use graceful shutdowns, other components rely on existence of this struct
// and if it's dropped, everything will start going offline
_task_manager: TaskManager,
packet_type: Option<PacketType>,
}
#[wasm_bindgen]
@@ -67,6 +69,7 @@ pub struct NymClientBuilder {
bandwidth_controller:
Option<BandwidthController<FakeClient<DirectSigningNyxdClient>, EphemeralStorage>>,
disabled_credentials: bool,
packet_type: Option<PacketType>,
}
#[wasm_bindgen]
@@ -82,6 +85,7 @@ impl NymClientBuilder {
on_message,
bandwidth_controller: None,
disabled_credentials: true,
packet_type: None,
}
}
@@ -132,6 +136,7 @@ impl NymClientBuilder {
on_message,
bandwidth_controller: None,
disabled_credentials: true,
packet_type: None,
}
}
@@ -189,6 +194,7 @@ impl NymClientBuilder {
client_state: Arc::new(started_client.client_state),
_full_topology: None,
_task_manager: started_client.task_manager,
packet_type: self.packet_type,
})
}
@@ -255,7 +261,7 @@ impl NymClient {
let input_msgs = request
.test_msgs
.into_iter()
.map(|p| InputMessage::new_regular(recipient, p, lane))
.map(|p| InputMessage::new_regular(recipient, p, lane, None))
.collect();
self.client_input.send_messages(input_msgs)
@@ -275,7 +281,7 @@ impl NymClient {
let lane = TransmissionLane::General;
let input_msg = InputMessage::new_regular(recipient, message, lane);
let input_msg = InputMessage::new_regular(recipient, message, lane, self.packet_type);
self.client_input.send_message(input_msg)
}
@@ -302,7 +308,8 @@ impl NymClient {
let lane = TransmissionLane::General;
let input_msg = InputMessage::new_anonymous(recipient, message, reply_surbs, lane);
let input_msg =
InputMessage::new_anonymous(recipient, message, reply_surbs, lane, self.packet_type);
self.client_input.send_message(input_msg)
}
@@ -320,7 +327,7 @@ impl NymClient {
let lane = TransmissionLane::General;
let input_msg = InputMessage::new_reply(sender_tag, message, lane);
let input_msg = InputMessage::new_reply(sender_tag, message, lane, self.packet_type);
self.client_input.send_message(input_msg)
}
}
common/client-core/src/client/base_client/mod.rs
+5 -5
View File
@@ -409,7 +409,7 @@ where
Ok(())
}
// controller for sending sphinx packets to mixnet (either real traffic or cover traffic)
// controller for sending packets to mixnet (either real traffic or cover traffic)
// TODO: if we want to send control messages to gateway_client, this CAN'T take the ownership
// over it. Perhaps GatewayClient needs to be thread-shareable or have some channel for
// requests?
@@ -524,11 +524,11 @@ where
task_manager.subscribe(),
);
// The sphinx_message_sender is the transmitter for any component generating sphinx packets
// The message_sender is the transmitter for any component generating sphinx packets
// that are to be sent to the mixnet. They are used by cover traffic stream and real
// traffic stream.
// The MixTrafficController then sends the actual traffic
let sphinx_message_sender =
let message_sender =
Self::start_mix_traffic_controller(gateway_client, task_manager.subscribe());
// Channels that the websocket listener can use to signal downstream to the real traffic
@@ -550,7 +550,7 @@ where
shared_topology_accessor.clone(),
ack_receiver,
input_receiver,
sphinx_message_sender.clone(),
message_sender.clone(),
reply_storage,
reply_controller_sender.clone(),
reply_controller_receiver,
@@ -569,7 +569,7 @@ where
self.key_manager.ack_key(),
self_address,
shared_topology_accessor.clone(),
sphinx_message_sender,
message_sender,
task_manager.subscribe(),
);
}
common/client-core/src/client/cover_traffic_stream.rs
+1 -1
View File
@@ -45,7 +45,7 @@ where
#[cfg(target_arch = "wasm32")]
next_delay: Pin<Box<wasm_timer::Delay>>,
/// Channel used for sending prepared sphinx packets to `MixTrafficController` that sends them
/// Channel used for sending prepared nym packets to `MixTrafficController` that sends them
/// out to the network without any further delays.
mix_tx: BatchMixMessageSender,
common/client-core/src/client/inbound_messages.rs
+51 -6
View File
@@ -4,6 +4,7 @@
use nym_sphinx::addressing::clients::Recipient;
use nym_sphinx::anonymous_replies::requests::AnonymousSenderTag;
use nym_sphinx::forwarding::packet::MixPacket;
use nym_sphinx::params::PacketType;
use nym_task::connections::TransmissionLane;
pub type InputMessageSender = tokio::sync::mpsc::Sender<InputMessage>;
@@ -53,18 +54,49 @@ pub enum InputMessage {
data: Vec<u8>,
lane: TransmissionLane,
},
MessageWrapper {
message: Box<InputMessage>,
packet_type: PacketType,
},
}
impl InputMessage {
pub fn new_premade(msgs: Vec<MixPacket>, lane: TransmissionLane) -> Self {
InputMessage::Premade { msgs, lane }
pub fn new_premade(
msgs: Vec<MixPacket>,
lane: TransmissionLane,
packet_type: PacketType,
) -> Self {
let message = InputMessage::Premade { msgs, lane };
if packet_type == PacketType::Mix {
message
} else {
InputMessage::new_wrapper(message, packet_type)
}
}
pub fn new_regular(recipient: Recipient, data: Vec<u8>, lane: TransmissionLane) -> Self {
InputMessage::Regular {
pub fn new_wrapper(message: InputMessage, packet_type: PacketType) -> Self {
InputMessage::MessageWrapper {
message: Box::new(message),
packet_type,
}
}
pub fn new_regular(
recipient: Recipient,
data: Vec<u8>,
lane: TransmissionLane,
packet_type: Option<PacketType>,
) -> Self {
let message = InputMessage::Regular {
recipient,
data,
lane,
};
if let Some(packet_type) = packet_type {
InputMessage::new_wrapper(message, packet_type)
} else {
message
}
}
@@ -73,12 +105,18 @@ impl InputMessage {
data: Vec<u8>,
reply_surbs: u32,
lane: TransmissionLane,
packet_type: Option<PacketType>,
) -> Self {
InputMessage::Anonymous {
let message = InputMessage::Anonymous {
recipient,
data,
reply_surbs,
lane,
};
if let Some(packet_type) = packet_type {
InputMessage::new_wrapper(message, packet_type)
} else {
message
}
}
@@ -86,11 +124,17 @@ impl InputMessage {
recipient_tag: AnonymousSenderTag,
data: Vec<u8>,
lane: TransmissionLane,
packet_type: Option<PacketType>,
) -> Self {
InputMessage::Reply {
let message = InputMessage::Reply {
recipient_tag,
data,
lane,
};
if let Some(packet_type) = packet_type {
InputMessage::new_wrapper(message, packet_type)
} else {
message
}
}
@@ -100,6 +144,7 @@ impl InputMessage {
| InputMessage::Anonymous { lane, .. }
| InputMessage::Reply { lane, .. }
| InputMessage::Premade { lane, .. } => lane,
InputMessage::MessageWrapper { message, .. } => message.lane(),
}
}
}
common/client-core/src/client/mix_traffic.rs
+3 -3
View File
@@ -39,15 +39,15 @@ where
pub fn new(
gateway_client: GatewayClient<C, St>,
) -> (MixTrafficController<C, St>, BatchMixMessageSender) {
let (sphinx_message_sender, sphinx_message_receiver) =
let (message_sender, message_receiver) =
tokio::sync::mpsc::channel(MIX_MESSAGE_RECEIVER_BUFFER_SIZE);
(
MixTrafficController {
gateway_client,
mix_rx: sphinx_message_receiver,
mix_rx: message_receiver,
consecutive_gateway_failure_count: 0,
},
sphinx_message_sender,
message_sender,
)
}
common/client-core/src/client/real_messages_control/acknowledgement_control/input_message_listener.rs
+44 -4
View File
@@ -9,6 +9,7 @@ use log::*;
use nym_sphinx::addressing::clients::Recipient;
use nym_sphinx::anonymous_replies::requests::AnonymousSenderTag;
use nym_sphinx::forwarding::packet::MixPacket;
use nym_sphinx::params::PacketType;
use nym_task::connections::TransmissionLane;
use rand::{CryptoRng, Rng};
@@ -71,10 +72,11 @@ where
recipient: Recipient,
content: Vec<u8>,
lane: TransmissionLane,
packet_type: PacketType,
) {
if let Err(err) = self
.message_handler
.try_send_plain_message(recipient, content, lane)
.try_send_plain_message(recipient, content, lane, packet_type)
.await
{
warn!("failed to send a plain message - {err}")
@@ -87,10 +89,11 @@ where
content: Vec<u8>,
reply_surbs: u32,
lane: TransmissionLane,
packet_type: PacketType,
) {
if let Err(err) = self
.message_handler
.try_send_message_with_reply_surbs(recipient, content, reply_surbs, lane)
.try_send_message_with_reply_surbs(recipient, content, reply_surbs, lane, packet_type)
.await
{
warn!("failed to send a repliable message - {err}")
@@ -103,14 +106,17 @@ where
recipient,
data,
lane,
} => self.handle_plain_message(recipient, data, lane).await,
} => {
self.handle_plain_message(recipient, data, lane, PacketType::Mix)
.await
}
InputMessage::Anonymous {
recipient,
data,
reply_surbs,
lane,
} => {
self.handle_repliable_message(recipient, data, reply_surbs, lane)
self.handle_repliable_message(recipient, data, reply_surbs, lane, PacketType::Mix)
.await
}
InputMessage::Reply {
@@ -121,6 +127,40 @@ where
self.handle_reply(recipient_tag, data, lane).await;
}
InputMessage::Premade { msgs, lane } => self.handle_premade_packets(msgs, lane).await,
InputMessage::MessageWrapper {
message,
packet_type,
} => match *message {
InputMessage::Regular {
recipient,
data,
lane,
} => {
self.handle_plain_message(recipient, data, lane, packet_type)
.await
}
InputMessage::Anonymous {
recipient,
data,
reply_surbs,
lane,
} => {
self.handle_repliable_message(recipient, data, reply_surbs, lane, packet_type)
.await
}
InputMessage::Reply {
recipient_tag,
data,
lane,
} => {
self.handle_reply(recipient_tag, data, lane).await;
}
InputMessage::Premade { msgs, lane } => {
self.handle_premade_packets(msgs, lane).await
}
// MessageWrappers can't be nested
InputMessage::MessageWrapper { .. } => unimplemented!(),
},
};
}
common/client-core/src/client/real_messages_control/acknowledgement_control/retransmission_request_listener.rs
+4 -2
View File
@@ -11,9 +11,9 @@ use crate::client::real_messages_control::real_traffic_stream::RealMessage;
use crate::client::replies::reply_controller::ReplyControllerSender;
use futures::StreamExt;
use log::*;
use nym_sphinx::addressing::clients::Recipient;
use nym_sphinx::chunking::fragment::Fragment;
use nym_sphinx::preparer::PreparedFragment;
use nym_sphinx::{addressing::clients::Recipient, params::PacketType};
use nym_task::connections::TransmissionLane;
use rand::{CryptoRng, Rng};
use std::sync::{Arc, Weak};
@@ -51,8 +51,10 @@ where
) -> Result<PreparedFragment, PreparationError> {
debug!("retransmitting normal packet...");
// TODO: Figure out retransmission packet type signaling
self.message_handler
.try_prepare_single_chunk_for_sending(packet_recipient, chunk_data)
.try_prepare_single_chunk_for_sending(packet_recipient, chunk_data, PacketType::Mix)
.await
}
common/client-core/src/client/real_messages_control/message_handler.rs
+17 -4
View File
@@ -15,7 +15,7 @@ use nym_sphinx::anonymous_replies::requests::{AnonymousSenderTag, RepliableMessa
use nym_sphinx::anonymous_replies::{ReplySurb, SurbEncryptionKey};
use nym_sphinx::chunking::fragment::{Fragment, FragmentIdentifier};
use nym_sphinx::message::NymMessage;
use nym_sphinx::params::{PacketSize, DEFAULT_NUM_MIX_HOPS};
use nym_sphinx::params::{PacketSize, PacketType, DEFAULT_NUM_MIX_HOPS};
use nym_sphinx::preparer::{MessagePreparer, PreparedFragment};
use nym_sphinx::Delay;
use nym_task::connections::TransmissionLane;
@@ -417,9 +417,10 @@ where
recipient: Recipient,
message: Vec<u8>,
lane: TransmissionLane,
packet_type: PacketType,
) -> Result<(), PreparationError> {
let message = NymMessage::new_plain(message);
self.try_split_and_send_non_reply_message(message, recipient, lane)
self.try_split_and_send_non_reply_message(message, recipient, lane, packet_type)
.await
}
@@ -428,6 +429,7 @@ where
message: NymMessage,
recipient: Recipient,
lane: TransmissionLane,
packet_type: PacketType,
) -> Result<(), PreparationError> {
// TODO: I really dislike existence of this assertion, it implies code has to be re-organised
debug_assert!(!matches!(message, NymMessage::Reply(_)));
@@ -453,6 +455,7 @@ where
topology,
&self.config.ack_key,
&recipient,
&packet_type,
)?;
let real_message = RealMessage::new(
@@ -476,6 +479,7 @@ where
&mut self,
recipient: Recipient,
amount: u32,
packet_type: PacketType,
) -> Result<(), PreparationError> {
let sender_tag = self.get_or_create_sender_tag(&recipient);
let (reply_surbs, reply_keys) =
@@ -490,6 +494,7 @@ where
message,
recipient,
TransmissionLane::AdditionalReplySurbs,
packet_type,
)
.await?;
@@ -505,6 +510,7 @@ where
message: Vec<u8>,
num_reply_surbs: u32,
lane: TransmissionLane,
packet_type: PacketType,
) -> Result<(), SurbWrappedPreparationError> {
let sender_tag = self.get_or_create_sender_tag(&recipient);
let (reply_surbs, reply_keys) = self
@@ -514,7 +520,7 @@ where
let message =
NymMessage::new_repliable(RepliableMessage::new_data(message, sender_tag, reply_surbs));
self.try_split_and_send_non_reply_message(message, recipient, lane)
self.try_split_and_send_non_reply_message(message, recipient, lane, packet_type)
.await?;
log::trace!("storing {} reply keys", reply_keys.len());
@@ -527,13 +533,20 @@ where
&mut self,
recipient: Recipient,
chunk: Fragment,
packet_type: PacketType,
) -> Result<PreparedFragment, PreparationError> {
let topology_permit = self.topology_access.get_read_permit().await;
let topology = self.get_topology(&topology_permit)?;
let prepared_fragment = self
.message_preparer
.prepare_chunk_for_sending(chunk, topology, &self.config.ack_key, &recipient)
.prepare_chunk_for_sending(
chunk,
topology,
&self.config.ack_key,
&recipient,
&packet_type,
)
.unwrap();
Ok(prepared_fragment)
common/client-core/src/client/real_messages_control/real_traffic_stream.rs
+2 -2
View File
@@ -92,7 +92,7 @@ where
// messages.
sending_delay_controller: SendingDelayController,
/// Channel used for sending prepared sphinx packets to `MixTrafficController` that sends them
/// Channel used for sending prepared packets to `MixTrafficController` that sends them
/// out to the network without any further delays.
mix_tx: BatchMixMessageSender,
@@ -386,7 +386,7 @@ where
// On every iteration we get new messages from upstream. Given that these come bunched
// in `Vec`, this ensures that on average we will fetch messages faster than we can
// send, which is a condition for being able to multiplex sphinx packets from multiple
// send, which is a condition for being able to multiplex packets from multiple
// data streams.
match Pin::new(&mut self.real_receiver).poll_recv(cx) {
// in the case our real message channel stream was closed, we should also indicate we are closed
common/client-core/src/client/replies/reply_controller/mod.rs
+5 -1
View File
@@ -512,7 +512,11 @@ where
let to_send = min(remaining, 100);
if let Err(err) = self
.message_handler
.try_send_additional_reply_surbs(recipient, to_send)
.try_send_additional_reply_surbs(
recipient,
to_send,
nym_sphinx::params::PacketType::Mix,
)
.await
{
warn!("failed to send additional surbs to {recipient} - {err}");
common/client-core/src/client/transmission_buffer.rs
+1 -1
View File
@@ -28,7 +28,7 @@ impl SizedData for RealMessage {
impl SizedData for Fragment {
fn data_size(&self) -> usize {
// note that raw `Fragment` is smaller than sphinx packet payload
// note that raw `Fragment` is smaller than packet payload
// as it doesn't include surb-ack or the [shared] key materials
self.payload_size()
}
common/node-tester-utils/Cargo.toml
+1
View File
@@ -17,6 +17,7 @@ tokio = { workspace = true, features = ["macros"]}
nym-crypto = { path = "../crypto", features = ["asymmetric"] }
nym-task = { path = "../task" }
nym-topology = { path = "../topology" }
nym-sphinx-params = { path = "../nymsphinx/params" }
# TODO: do we need the whole nymsphinx?
nym-sphinx = { path = "../nymsphinx" }
common/node-tester-utils/src/tester.rs
+9 -1
View File
@@ -10,6 +10,7 @@ use nym_sphinx::addressing::clients::Recipient;
use nym_sphinx::message::NymMessage;
use nym_sphinx::params::{PacketSize, DEFAULT_NUM_MIX_HOPS};
use nym_sphinx::preparer::{FragmentPreparer, PreparedFragment};
use nym_sphinx_params::PacketType;
use nym_topology::{gateway, mix, NymTopology};
use rand::{CryptoRng, Rng};
use serde::Serialize;
@@ -170,7 +171,14 @@ where
// TODO: can we avoid this arc clone?
let ack_key = Arc::clone(&self.ack_key);
Ok(self.prepare_chunk_for_sending(fragment, topology, &ack_key, &address, &address)?)
Ok(self.prepare_chunk_for_sending(
fragment,
topology,
&ack_key,
&address,
&address,
&PacketType::Mix,
)?)
}
}
common/nymsphinx/framing/src/codec.rs
+1
View File
@@ -84,6 +84,7 @@ impl Decoder for NymCodec {
match header.packet_type {
PacketType::Outfox => NymPacket::outfox_from_bytes(slice)?,
PacketType::Mix => NymPacket::sphinx_from_bytes(slice)?,
PacketType::Vpn => NymPacket::sphinx_from_bytes(slice)?,
}
} else {
return Ok(None);
common/nymsphinx/params/src/packet_types.rs
+3
View File
@@ -19,6 +19,9 @@ pub enum PacketType {
#[default]
Mix = 0,
/// Represents a packet that should be sent through the network as fast as possible.
Vpn = 1,
/// Abusing this to add Outfox support
Outfox = 2,
}
common/nymsphinx/src/preparer/mod.rs
+25 -8
View File
@@ -5,6 +5,7 @@ use crate::message::{NymMessage, ACK_OVERHEAD};
use crate::NymsphinxPayloadBuilder;
use nym_crypto::asymmetric::encryption;
use nym_crypto::Digest;
use nym_outfox::packet::OutfoxPacket;
use nym_sphinx_acknowledgements::surb_ack::SurbAck;
use nym_sphinx_acknowledgements::AckKey;
use nym_sphinx_addressing::clients::Recipient;
@@ -13,7 +14,7 @@ use nym_sphinx_anonymous_replies::reply_surb::ReplySurb;
use nym_sphinx_chunking::fragment::{Fragment, FragmentIdentifier};
use nym_sphinx_forwarding::packet::MixPacket;
use nym_sphinx_params::packet_sizes::PacketSize;
use nym_sphinx_params::{ReplySurbKeyDigestAlgorithm, DEFAULT_NUM_MIX_HOPS};
use nym_sphinx_params::{PacketType, ReplySurbKeyDigestAlgorithm, DEFAULT_NUM_MIX_HOPS};
use nym_sphinx_types::builder::SphinxPacketBuilder;
use nym_sphinx_types::{delays, Delay, NymPacket};
use nym_topology::{NymTopology, NymTopologyError};
@@ -170,6 +171,7 @@ pub trait FragmentPreparer {
ack_key: &AckKey,
packet_sender: &Recipient,
packet_recipient: &Recipient,
packet_type: &PacketType,
) -> Result<PreparedFragment, NymTopologyError> {
// each plain or repliable packet (i.e. not a reply) attaches an ephemeral public key so that the recipient
// could perform diffie-hellman with its own keys followed by a kdf to re-derive
@@ -208,12 +210,25 @@ pub trait FragmentPreparer {
// create the actual sphinx packet here. With valid route and correct payload size,
// there's absolutely no reason for this call to fail.
let sphinx_packet = NymPacket::Sphinx(
SphinxPacketBuilder::new()
.with_payload_size(packet_size.payload_size())
.build_packet(packet_payload, &route, &destination, &delays)
.unwrap(),
);
let packet = match packet_type {
PacketType::Outfox => NymPacket::Outfox(OutfoxPacket::build(
packet_payload,
route.as_slice().try_into()?,
Some(packet_size.payload_size()),
)?),
PacketType::Mix => NymPacket::Sphinx(
SphinxPacketBuilder::new()
.with_payload_size(packet_size.payload_size())
.build_packet(packet_payload, &route, &destination, &delays)
.unwrap(),
),
PacketType::Vpn => NymPacket::Sphinx(
SphinxPacketBuilder::new()
.with_payload_size(packet_size.payload_size())
.build_packet(packet_payload, &route, &destination, &delays)
.unwrap(),
),
};
// from the previously constructed route extract the first hop
let first_hop_address =
@@ -224,7 +239,7 @@ pub trait FragmentPreparer {
// well as the total delay of the ack packet.
// note that the last hop of the packet is a gateway that does not do any delays
total_delay: delays.iter().take(delays.len() - 1).sum::<Delay>() + ack_delay,
mix_packet: MixPacket::new(first_hop_address, sphinx_packet, Default::default()),
mix_packet: MixPacket::new(first_hop_address, packet, Default::default()),
fragment_identifier,
})
}
@@ -335,6 +350,7 @@ where
topology: &NymTopology,
ack_key: &AckKey,
packet_recipient: &Recipient,
packet_type: &PacketType,
) -> Result<PreparedFragment, NymTopologyError> {
let sender = self.sender_address;
@@ -345,6 +361,7 @@ where
ack_key,
&sender,
packet_recipient,
packet_type,
)
}
common/nymsphinx/types/src/lib.rs
+1 -1
View File
@@ -1,7 +1,7 @@
// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
pub use nym_outfox::{format::MIX_PARAMS_LEN, packet::OUTFOX_PACKET_OVERHEAD};
pub use nym_outfox::{error::OutfoxError, format::MIX_PARAMS_LEN, packet::OUTFOX_PACKET_OVERHEAD};
// re-exporting types and constants available in sphinx
use nym_outfox::packet::OutfoxPacket;
pub use sphinx_packet::{
common/socks5-client-core/src/socks/client.rs
+14 -1
View File
@@ -18,6 +18,7 @@ use nym_socks5_requests::{
};
use nym_sphinx::addressing::clients::Recipient;
use nym_sphinx::params::PacketSize;
use nym_sphinx::params::PacketType;
use nym_task::connections::{LaneQueueLengths, TransmissionLane};
use nym_task::TaskClient;
use pin_project::pin_project;
@@ -185,6 +186,7 @@ pub(crate) struct SocksClient {
started_proxy: bool,
lane_queue_lengths: LaneQueueLengths,
shutdown_listener: TaskClient,
packet_type: Option<PacketType>,
}
impl Drop for SocksClient {
@@ -213,6 +215,7 @@ impl SocksClient {
self_address: &Recipient,
lane_queue_lengths: LaneQueueLengths,
mut shutdown_listener: TaskClient,
packet_type: Option<PacketType>,
) -> Self {
// If this task fails and exits, we don't want to send shutdown signal
shutdown_listener.mark_as_success();
@@ -233,6 +236,7 @@ impl SocksClient {
started_proxy: false,
lane_queue_lengths,
shutdown_listener,
packet_type,
}
}
@@ -349,6 +353,7 @@ impl SocksClient {
msg.into_bytes(),
self.config.connection_start_surbs,
TransmissionLane::ConnectionId(self.connection_id),
self.packet_type,
);
self.input_sender
.send(input_message)
@@ -371,6 +376,7 @@ impl SocksClient {
self.service_provider,
msg.into_bytes(),
TransmissionLane::ConnectionId(self.connection_id),
self.packet_type,
);
self.input_sender
.send(input_message)
@@ -408,6 +414,7 @@ impl SocksClient {
let request_version = self.config.request_version();
let recipient = self.service_provider;
let packet_type = self.packet_type;
let (stream, _) = ProxyRunner::new(
stream,
local_stream_remote,
@@ -439,9 +446,15 @@ impl SocksClient {
provider_message.into_bytes(),
per_request_surbs,
lane,
packet_type,
)
} else {
InputMessage::new_regular(recipient, provider_message.into_bytes(), lane)
InputMessage::new_regular(
recipient,
provider_message.into_bytes(),
lane,
packet_type,
)
}
})
.await
common/socks5-client-core/src/socks/server.rs
+1
View File
@@ -104,6 +104,7 @@ impl SphinxSocksServer {
&self.self_address,
self.lane_queue_lengths.clone(),
self.shutdown.clone(),
None
);
tokio::spawn(async move {
common/topology/src/lib.rs
+7
View File
@@ -8,6 +8,7 @@ use nym_mixnet_contract_common::{GatewayBond, IdentityKeyRef, MixId};
use nym_sphinx_addressing::nodes::NodeIdentity;
use nym_sphinx_types::Node as SphinxNode;
use rand::{CryptoRng, Rng};
use std::array::TryFromSliceError;
use std::collections::BTreeMap;
use std::convert::TryInto;
use std::fmt::{self, Display, Formatter};
@@ -55,6 +56,12 @@ pub enum NymTopologyError {
// We can't import SurbAckRecoveryError due to cyclic dependency, this is a bit dirty
#[error("Could not build payload")]
PayloadBuilder,
#[error("Outfox: {0}")]
Outfox(#[from] nym_sphinx_types::OutfoxError),
#[error("{0}")]
FromSlice(#[from] TryFromSliceError),
}
#[derive(Debug, Clone)]
contracts/Cargo.lock
Generated
+2 -2
View File
@@ -1676,9 +1676,9 @@ dependencies = [
[[package]]
name = "rustix"
version = "0.37.14"
version = "0.37.15"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d9b864d3c18a5785a05953adeed93e2dca37ed30f18e69bba9f30079d51f363f"
checksum = "a0661814f891c57c930a610266415528da53c4933e6dea5fb350cbfe048a9ece"
dependencies = [
"bitflags",
"errno",
mixnode/src/main.rs
+6 -1
View File
@@ -51,9 +51,14 @@ fn test_function() {
async fn main() {
cfg_if::cfg_if! {
if #[cfg(feature = "cpucycles")] {
setup_tracing!("/tmp/tracing.log");
let home_dir = dirs::home_dir().expect("Could not get $HOME");
let logs_dir = home_dir.join(".nym").join("logs");
let logs_dir_str = logs_dir.to_str().expect("Could not construct logs path");
setup_tracing!(logs_dir_str);
info!("CPU cycles measurement is ON")
} else {
setup_logging();
info!("CPU cycles measurement is OFF")
}
}
nym-api/src/network_monitor/chunker.rs
+11 -3
View File
@@ -3,7 +3,7 @@
use nym_sphinx::forwarding::packet::MixPacket;
use nym_sphinx::message::NymMessage;
use nym_sphinx::params::PacketSize;
use nym_sphinx::params::{PacketSize, PacketType};
use nym_sphinx::{
acknowledgements::AckKey, addressing::clients::Recipient, preparer::MessagePreparer,
};
@@ -43,12 +43,13 @@ impl Chunker {
message: Vec<u8>,
topology: &NymTopology,
packet_sender: Recipient,
packet_type: PacketType,
) -> Vec<MixPacket> {
// I really dislike how we have to overwrite the parameter of the `MessagePreparer` on each run
// but without some significant API changes in the `MessagePreparer` this was the easiest
// way to being able to have variable sender address.
self.message_preparer.set_sender_address(packet_sender);
self.prepare_packets(message, topology, packet_sender)
self.prepare_packets(message, topology, packet_sender, packet_type)
}
fn prepare_packets(
@@ -56,6 +57,7 @@ impl Chunker {
message: Vec<u8>,
topology: &NymTopology,
packet_sender: Recipient,
packet_type: PacketType,
) -> Vec<MixPacket> {
let ack_key: AckKey = AckKey::new(&mut self.rng);
@@ -68,7 +70,13 @@ impl Chunker {
// don't bother with acks etc. for time being
let prepared_fragment = self
.message_preparer
.prepare_chunk_for_sending(message_chunk, topology, &ack_key, &packet_sender)
.prepare_chunk_for_sending(
message_chunk,
topology,
&ack_key,
&packet_sender,
&packet_type,
)
.unwrap();
mix_packets.push(prepared_fragment.mix_packet);
nym-api/src/network_monitor/mod.rs
+2
View File
@@ -20,6 +20,7 @@ use futures::channel::mpsc;
use nym_bandwidth_controller::BandwidthController;
use nym_credential_storage::persistent_storage::PersistentStorage;
use nym_crypto::asymmetric::{encryption, identity};
use nym_sphinx::params::PacketType;
use nym_sphinx::receiver::MessageReceiver;
use nym_task::TaskManager;
use std::sync::Arc;
@@ -132,6 +133,7 @@ impl<'a> NetworkMonitorBuilder<'a> {
received_processor,
summary_producer,
self.node_status_storage,
PacketType::Mix,
);
NetworkMonitorRunnables {
nym-api/src/network_monitor/monitor/mod.rs
+11 -3
View File
@@ -10,6 +10,7 @@ use crate::network_monitor::test_route::TestRoute;
use crate::storage::NymApiStorage;
use crate::support::config::Config;
use log::{debug, error, info};
use nym_sphinx::params::PacketType;
use nym_sphinx::receiver::MessageReceiver;
use nym_task::TaskClient;
use std::collections::{HashMap, HashSet};
@@ -44,6 +45,8 @@ pub(super) struct Monitor<R: MessageReceiver + Send + 'static> {
/// The minimum number of test routes that need to be constructed (and working) in order for
/// a monitor test run to be valid.
minimum_test_routes: usize,
packet_type: PacketType,
}
impl<R: MessageReceiver + Send> Monitor<R> {
@@ -54,6 +57,7 @@ impl<R: MessageReceiver + Send> Monitor<R> {
received_processor: ReceivedProcessor<R>,
summary_producer: SummaryProducer,
node_status_storage: NymApiStorage,
packet_type: PacketType,
) -> Self {
Monitor {
test_nonce: 1,
@@ -68,6 +72,7 @@ impl<R: MessageReceiver + Send> Monitor<R> {
route_test_packets: config.get_route_test_packets(),
test_routes: config.get_test_routes(),
minimum_test_routes: config.get_minimum_test_routes(),
packet_type,
}
}
@@ -122,8 +127,11 @@ impl<R: MessageReceiver + Send> Monitor<R> {
for route in routes {
let mut packet_preparer = self.packet_preparer.clone();
let route = route.clone();
let gateway_packets = packet_preparer
.prepare_test_route_viability_packets(&route, self.route_test_packets);
let gateway_packets = packet_preparer.prepare_test_route_viability_packets(
&route,
self.route_test_packets,
self.packet_type,
);
packets.push(gateway_packets);
}
@@ -230,7 +238,7 @@ impl<R: MessageReceiver + Send> Monitor<R> {
info!("Generating test mix packets for all the network nodes...");
let prepared_packets = self
.packet_preparer
.prepare_test_packets(self.test_nonce, routes)
.prepare_test_packets(self.test_nonce, routes, self.packet_type)
.await;
let total_sent = prepared_packets
nym-api/src/network_monitor/monitor/preparer.rs
+11 -3
View File
@@ -11,6 +11,7 @@ use nym_crypto::asymmetric::{encryption, identity};
use nym_mixnet_contract_common::{Addr, GatewayBond, Layer, MixId, MixNodeBond};
use nym_sphinx::addressing::clients::Recipient;
use nym_sphinx::forwarding::packet::MixPacket;
use nym_sphinx::params::PacketType;
use nym_topology::{gateway, mix, NymTopology};
use rand::seq::SliceRandom;
use rand::{thread_rng, Rng};
@@ -170,6 +171,7 @@ impl PacketPreparer {
packet: &TestPacket,
topology: &NymTopology,
packet_recipient: Recipient,
packet_type: PacketType,
) -> MixPacket {
// this should be done only once. We can't really do it at construction time
// as there's no sane Default for Recipient
@@ -180,6 +182,7 @@ impl PacketPreparer {
packet.to_bytes(),
topology,
packet_recipient,
packet_type,
);
assert_eq!(
mix_packets.len(),
@@ -368,12 +371,14 @@ impl PacketPreparer {
&mut self,
route: &TestRoute,
num: usize,
packet_type: PacketType,
) -> GatewayPackets {
let mut mix_packets = Vec::with_capacity(num);
let test_packet = route.self_test_packet();
let recipient = self.create_packet_sender(route.gateway());
for _ in 0..num {
let mix_packet = self.wrap_test_packet(&test_packet, route.topology(), recipient);
let mix_packet =
self.wrap_test_packet(&test_packet, route.topology(), recipient, packet_type);
mix_packets.push(mix_packet)
}
@@ -428,6 +433,7 @@ impl PacketPreparer {
&mut self,
test_nonce: u64,
test_routes: &[TestRoute],
packet_type: PacketType,
) -> PreparedPackets {
// only test mixnodes that are rewarded, i.e. that will be rewarded in this interval.
// (remember that "idle" nodes are still part of that set)
@@ -462,7 +468,8 @@ impl PacketPreparer {
let topology = test_route.substitute_mix(mixnode);
// produce n mix packets
for _ in 0..self.per_node_test_packets {
let mix_packet = self.wrap_test_packet(&test_packet, &topology, recipient);
let mix_packet =
self.wrap_test_packet(&test_packet, &topology, recipient, packet_type);
mix_packets.push(mix_packet);
}
}
@@ -482,7 +489,8 @@ impl PacketPreparer {
let topology = test_route.substitute_gateway(gateway);
// produce n mix packets
for _ in 0..self.per_node_test_packets {
let mix_packet = self.wrap_test_packet(&test_packet, &topology, recipient);
let mix_packet =
self.wrap_test_packet(&test_packet, &topology, recipient, packet_type);
gateway_mix_packets.push(mix_packet);
}
nym-outfox/src/error.rs
+1 -1
View File
@@ -5,7 +5,7 @@ use crate::lion::MIN_MESSAGE_LEN;
use chacha20::cipher::InvalidLength;
use thiserror::Error;
#[derive(Debug, Error)]
#[derive(Debug, Error, Clone)]
pub enum OutfoxError {
#[error("Lengths mismatch, expected: {expected}, got: {got}")]
LenMismatch { expected: usize, got: usize },
nym-outfox/src/format.rs
+2 -2
View File
@@ -92,7 +92,7 @@ use std::convert::TryFrom;
pub struct MixCreationParameters {
/// The routing length is inner first, so \[0\] is the innermost routing length, etc (in bytes)
/// In our stratified topology this will always be 3
pub routing_information_length_by_stage: [u8; 3],
pub routing_information_length_by_stage: [u8; 4],
/// The payload length (in bytes)
pub payload_length_bytes: u16,
}
@@ -127,7 +127,7 @@ impl MixCreationParameters {
/// Create a set of parameters for a mix packet format.
pub fn new(payload_length_bytes: u16) -> MixCreationParameters {
MixCreationParameters {
routing_information_length_by_stage: [DEFAULT_ROUTING_INFO_SIZE; 3],
routing_information_length_by_stage: [DEFAULT_ROUTING_INFO_SIZE; 4],
payload_length_bytes,
}
}
nym-outfox/src/packet.rs
+1 -1
View File
@@ -51,7 +51,7 @@ impl OutfoxPacket {
pub fn build<M: AsRef<[u8]>>(
payload: M,
route: &[Node; 3],
route: &[Node; 4],
packet_size: Option<usize>,
) -> Result<OutfoxPacket, OutfoxError> {
let mut secret_key = [0; 32];
sdk/rust/nym-sdk/src/mixnet/client.rs
+1
View File
@@ -592,6 +592,7 @@ where
client_state,
reconstructed_receiver,
task_manager: started_client.task_manager,
packet_type: None,
})
}
}
sdk/rust/nym-sdk/src/mixnet/native_client.rs
+6 -2
View File
@@ -6,6 +6,7 @@ use nym_client_core::client::{
};
use nym_sphinx::{
addressing::clients::{ClientIdentity, Recipient},
params::PacketType,
receiver::ReconstructedMessage,
};
use nym_task::{
@@ -45,6 +46,7 @@ pub struct MixnetClient {
/// The task manager that controlls all the spawned tasks that the clients uses to do it's job.
pub(crate) task_manager: TaskManager,
pub(crate) packet_type: Option<PacketType>,
}
impl MixnetClient {
@@ -163,9 +165,11 @@ impl MixnetClient {
let lane = TransmissionLane::General;
let input_msg = match surbs {
IncludedSurbs::Amount(surbs) => {
InputMessage::new_anonymous(address, message, surbs, lane)
InputMessage::new_anonymous(address, message, surbs, lane, self.packet_type)
}
IncludedSurbs::ExposeSelfAddress => {
InputMessage::new_regular(address, message, lane, self.packet_type)
}
IncludedSurbs::ExposeSelfAddress => InputMessage::new_regular(address, message, lane),
};
self.send(input_msg).await
}