From 8824a875d9fd2852d551fd31180ddffcc26842e6 Mon Sep 17 00:00:00 2001 From: Simon Wicky Date: Wed, 3 Apr 2024 17:14:40 +0200 Subject: [PATCH] drop cover traffic --- .../client-core/src/client/base_client/mod.rs | 10 ++ .../src/client/real_messages_control/mod.rs | 2 + .../real_traffic_stream.rs | 75 +++++++--- .../client-core/src/client/received_buffer.rs | 10 ++ common/nymsphinx/cover/src/lib.rs | 128 ++++++++++++++++++ .../receiver/connection_handler.rs | 10 +- 6 files changed, 213 insertions(+), 22 deletions(-) diff --git a/common/client-core/src/client/base_client/mod.rs b/common/client-core/src/client/base_client/mod.rs index 47b119261a..d9da1aff69 100644 --- a/common/client-core/src/client/base_client/mod.rs +++ b/common/client-core/src/client/base_client/mod.rs @@ -290,6 +290,7 @@ where shutdown: TaskClient, packet_type: PacketType, stats_tx: PacketStatisticsReporter, + counter_receiver: mpsc::Receiver, ) { info!("Starting real traffic stream..."); @@ -305,12 +306,14 @@ where lane_queue_lengths, client_connection_rx, stats_tx, + counter_receiver, ) .start_with_shutdown(shutdown, packet_type); } // buffer controlling all messages fetched from provider // required so that other components would be able to use them (say the websocket) + #[allow(clippy::too_many_arguments)] fn start_received_messages_buffer_controller( local_encryption_keypair: Arc, query_receiver: ReceivedBufferRequestReceiver, @@ -319,6 +322,7 @@ where reply_controller_sender: ReplyControllerSender, shutdown: TaskClient, packet_statistics_control: PacketStatisticsReporter, + counter_sender: mpsc::Sender, ) { info!("Starting received messages buffer controller..."); let controller: ReceivedMessagesBufferController = @@ -329,6 +333,7 @@ where reply_key_storage, reply_controller_sender, packet_statistics_control, + counter_sender, ); controller.start_with_shutdown(shutdown) } @@ -606,6 +611,9 @@ where // channels responsible for controlling real messages let (input_sender, input_receiver) = tokio::sync::mpsc::channel::(1); + // drop cover traffic counter channel + let (counter_sender, counter_receiver) = mpsc::channel::(10000); + // channels responsible for controlling ack messages let (ack_sender, ack_receiver) = mpsc::unbounded(); let shared_topology_accessor = TopologyAccessor::new(); @@ -681,6 +689,7 @@ where reply_controller_sender.clone(), shutdown.fork("received_messages_buffer"), packet_stats_reporter.clone(), + counter_sender, ); // The message_sender is the transmitter for any component generating sphinx packets @@ -720,6 +729,7 @@ where shutdown.fork("real_traffic_controller"), self.config.debug.traffic.packet_type, packet_stats_reporter.clone(), + counter_receiver, ); if !self diff --git a/common/client-core/src/client/real_messages_control/mod.rs b/common/client-core/src/client/real_messages_control/mod.rs index d4914f1440..28dcd7ebed 100644 --- a/common/client-core/src/client/real_messages_control/mod.rs +++ b/common/client-core/src/client/real_messages_control/mod.rs @@ -146,6 +146,7 @@ impl RealMessagesController { lane_queue_lengths: LaneQueueLengths, client_connection_rx: ConnectionCommandReceiver, stats_tx: PacketStatisticsReporter, + counter_receiver: mpsc::Receiver, ) -> Self { let rng = OsRng; @@ -204,6 +205,7 @@ impl RealMessagesController { lane_queue_lengths, client_connection_rx, stats_tx, + counter_receiver, ); RealMessagesController { diff --git a/common/client-core/src/client/real_messages_control/real_traffic_stream.rs b/common/client-core/src/client/real_messages_control/real_traffic_stream.rs index ef91208f1b..30587c8124 100644 --- a/common/client-core/src/client/real_messages_control/real_traffic_stream.rs +++ b/common/client-core/src/client/real_messages_control/real_traffic_stream.rs @@ -8,13 +8,14 @@ use crate::client::real_messages_control::acknowledgement_control::SentPacketNot use crate::client::topology_control::TopologyAccessor; use crate::client::transmission_buffer::TransmissionBuffer; use crate::config; +use futures::channel::mpsc; use futures::task::{Context, Poll}; use futures::{Future, Stream, StreamExt}; use log::*; use nym_sphinx::acknowledgements::AckKey; use nym_sphinx::addressing::clients::Recipient; use nym_sphinx::chunking::fragment::FragmentIdentifier; -use nym_sphinx::cover::generate_loop_cover_packet; +use nym_sphinx::cover::{generate_drop_cover_packet, generate_loop_cover_packet}; use nym_sphinx::forwarding::packet::MixPacket; use nym_sphinx::params::PacketSize; use nym_sphinx::preparer::PreparedFragment; @@ -117,6 +118,8 @@ where /// Channel used for sending statistics events to `PacketStatisticsControl`. stats_tx: PacketStatisticsReporter, + //counter for drop cover traffic + counter_receiver: mpsc::Receiver, } #[derive(Debug)] @@ -155,7 +158,7 @@ pub(crate) type BatchRealMessageSender = type BatchRealMessageReceiver = tokio::sync::mpsc::Receiver<(Vec, TransmissionLane)>; pub(crate) enum StreamMessage { - Cover, + Cover(bool), Real(Box), } @@ -176,6 +179,7 @@ where lane_queue_lengths: LaneQueueLengths, client_connection_rx: ConnectionCommandReceiver, stats_tx: PacketStatisticsReporter, + counter_receiver: mpsc::Receiver, ) -> Self { OutQueueControl { config, @@ -190,6 +194,7 @@ where client_connection_rx, lane_queue_lengths, stats_tx, + counter_receiver, } } @@ -221,7 +226,7 @@ where trace!("created new message"); let (next_message, fragment_id, packet_size) = match next_message { - StreamMessage::Cover => { + StreamMessage::Cover(drop) => { let cover_traffic_packet_size = self.loop_cover_message_size(); trace!("the next loop cover message will be put in a {cover_traffic_packet_size} packet"); @@ -240,24 +245,44 @@ where return; } }; - - ( - generate_loop_cover_packet( - &mut self.rng, - topology_ref, - &self.config.ack_key, - &self.config.our_full_destination, - self.config.average_ack_delay, - self.config.traffic.average_packet_delay, - cover_traffic_packet_size, - self.config.traffic.packet_type, + if drop { + debug!("Sending a drop cover message"); + ( + generate_drop_cover_packet( + &mut self.rng, + topology_ref, + &self.config.ack_key, + &self.config.our_full_destination, + self.config.average_ack_delay, + self.config.traffic.average_packet_delay, + cover_traffic_packet_size, + self.config.traffic.packet_type, + ) + .expect( + "Somehow failed to generate a drop cover message with a valid topology", + ), + None, + cover_traffic_packet_size.size(), ) - .expect( - "Somehow failed to generate a loop cover message with a valid topology", - ), - None, - cover_traffic_packet_size.size(), - ) + } else { + ( + generate_loop_cover_packet( + &mut self.rng, + topology_ref, + &self.config.ack_key, + &self.config.our_full_destination, + self.config.average_ack_delay, + self.config.traffic.average_packet_delay, + cover_traffic_packet_size, + self.config.traffic.packet_type, + ) + .expect( + "Somehow failed to generate a loop cover message with a valid topology", + ), + None, + cover_traffic_packet_size.size(), + ) + } } StreamMessage::Real(real_message) => { let packet_size = real_message.packet_size(); @@ -385,6 +410,10 @@ where } fn poll_poisson(&mut self, cx: &mut Context<'_>) -> Poll> { + //For instantaneous drop cover traffic + // if let Ok(_) = self.counter_receiver.try_next() { + // return Poll::Ready(Some(StreamMessage::Cover(true))); + // }; // The average delay could change depending on if backpressure in the downstream channel // (mix_tx) was detected. self.adjust_current_average_message_sending_delay(); @@ -419,6 +448,7 @@ where // 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 packets from multiple // data streams. + let need_drop = self.counter_receiver.try_next(); 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 // (and whoever is using the stream should panic) @@ -438,7 +468,10 @@ where Poll::Ready(Some(StreamMessage::Real(Box::new(real_next)))) } else { // otherwise construct a dummy one - Poll::Ready(Some(StreamMessage::Cover)) + match need_drop { + Ok(_) => Poll::Ready(Some(StreamMessage::Cover(true))), + _ => Poll::Ready(Some(StreamMessage::Cover(false))), + } } } } diff --git a/common/client-core/src/client/received_buffer.rs b/common/client-core/src/client/received_buffer.rs index 021fd75d1a..8a71e933a8 100644 --- a/common/client-core/src/client/received_buffer.rs +++ b/common/client-core/src/client/received_buffer.rs @@ -47,6 +47,7 @@ struct ReceivedMessagesBufferInner { recently_reconstructed: HashSet, stats_tx: PacketStatisticsReporter, + counter_sender: mpsc::Sender, } impl ReceivedMessagesBufferInner { @@ -71,6 +72,11 @@ impl ReceivedMessagesBufferInner { .report(PacketStatisticsEvent::RealPacketReceived( fragment_data_size, )); + if self.counter_sender.try_send(1).is_err() { + { + log::debug!("Failed to send counter message"); + } + }; let fragment = match self.message_receiver.recover_fragment(fragment_data) { Err(err) => { @@ -164,6 +170,7 @@ impl ReceivedMessagesBuffer { reply_key_storage: SentReplyKeys, reply_controller_sender: ReplyControllerSender, stats_tx: PacketStatisticsReporter, + counter_sender: mpsc::Sender, ) -> Self { ReceivedMessagesBuffer { inner: Arc::new(Mutex::new(ReceivedMessagesBufferInner { @@ -173,6 +180,7 @@ impl ReceivedMessagesBuffer { message_sender: None, recently_reconstructed: HashSet::new(), stats_tx, + counter_sender, })), reply_key_storage, reply_controller_sender, @@ -505,12 +513,14 @@ impl ReceivedMessagesBufferControll reply_key_storage: SentReplyKeys, reply_controller_sender: ReplyControllerSender, packet_statistics_reporter: PacketStatisticsReporter, + counter_sender: mpsc::Sender, ) -> Self { let received_buffer = ReceivedMessagesBuffer::new( local_encryption_keypair, reply_key_storage, reply_controller_sender, packet_statistics_reporter, + counter_sender, ); ReceivedMessagesBufferController { diff --git a/common/nymsphinx/cover/src/lib.rs b/common/nymsphinx/cover/src/lib.rs index 66142a4cee..3770e5d834 100644 --- a/common/nymsphinx/cover/src/lib.rs +++ b/common/nymsphinx/cover/src/lib.rs @@ -1,6 +1,7 @@ // Copyright 2021 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 +use nym_crypto::asymmetric::{encryption, identity}; use nym_crypto::shared_key::new_ephemeral_shared_key; use nym_crypto::symmetric::stream_cipher; use nym_sphinx_acknowledgements::surb_ack::{SurbAck, SurbAckRecoveryError}; @@ -154,6 +155,133 @@ where Ok(MixPacket::new(first_hop_address, packet, packet_type)) } +#[allow(clippy::too_many_arguments)] +pub fn generate_drop_cover_packet( + rng: &mut R, + topology: &NymTopology, + ack_key: &AckKey, + full_address: &Recipient, + average_ack_delay: time::Duration, + average_packet_delay: time::Duration, + packet_size: PacketSize, + packet_type: PacketType, +) -> Result +where + R: RngCore + CryptoRng, +{ + // we don't care about total ack delay - we will not be retransmitting it anyway + let (_, ack_bytes) = generate_loop_cover_surb_ack( + rng, + topology, + ack_key, + full_address, + average_ack_delay, + packet_type, + )? + .prepare_for_sending()?; + + // cover message can't be distinguishable from a normal traffic so we have to go through + // all the effort of key generation, encryption, etc. Note here we are generating shared key + // with ourselves! + let (ephemeral_keypair, shared_key) = new_ephemeral_shared_key::< + PacketEncryptionAlgorithm, + PacketHkdfAlgorithm, + _, + >(rng, full_address.encryption_key()); + + let public_key_bytes = ephemeral_keypair.public_key().to_bytes(); + let cover_size = packet_size.plaintext_size() - public_key_bytes.len() - ack_bytes.len(); + + let mut cover_content: Vec<_> = LOOP_COVER_MESSAGE_PAYLOAD + .iter() + .cloned() + .chain(std::iter::once(1)) + .chain(std::iter::repeat(0)) + .take(cover_size) + .collect(); + + let zero_iv = stream_cipher::zero_iv::(); + stream_cipher::encrypt_in_place::( + &shared_key, + &zero_iv, + &mut cover_content, + ); + + // combine it together as follows: + // SURB_ACK_FIRST_HOP || SURB_ACK_DATA || EPHEMERAL_KEY || COVER_CONTENT + // (note: surb_ack_bytes contains SURB_ACK_FIRST_HOP || SURB_ACK_DATA ) + let packet_payload: Vec<_> = ack_bytes + .into_iter() + .chain(ephemeral_keypair.public_key().to_bytes().iter().cloned()) + .chain(cover_content) + .collect(); + + let route = + topology.random_route_to_gateway(rng, DEFAULT_NUM_MIX_HOPS, full_address.gateway())?; + let delays = nym_sphinx_routing::generate_hop_delays(average_packet_delay, route.len()); + + let client_id_pair = identity::KeyPair::from_bytes( + &[ + 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, + 0, 0, 0, + ], + &[ + 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, + 0, 0, 0, + ], + ) + .unwrap(); + let client_enc_pair = encryption::KeyPair::from_bytes( + &[ + 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, + 0, 0, 0, + ], + &[ + 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, + 0, 0, 0, + ], + ) + .unwrap(); + + let recipient = Recipient::new( + *client_id_pair.public_key(), + *client_enc_pair.public_key(), + *full_address.gateway(), + ); + + let destination = recipient.as_sphinx_destination(); + + let first_hop_address = + NymNodeRoutingAddress::try_from(route.first().unwrap().address).unwrap(); + + // once merged, that's an easy rng injection point for sphinx packets : ) + let packet = match packet_type { + PacketType::Mix => NymPacket::sphinx_build( + packet_size.payload_size(), + packet_payload, + &route, + &destination, + &delays, + )?, + #[allow(deprecated)] + PacketType::Vpn => NymPacket::sphinx_build( + packet_size.payload_size(), + packet_payload, + &route, + &destination, + &delays, + )?, + PacketType::Outfox => NymPacket::outfox_build( + packet_payload, + &route, + &destination, + Some(packet_size.plaintext_size()), + )?, + }; + + Ok(MixPacket::new(first_hop_address, packet, packet_type)) +} + /// Helper function used to determine if given message represents a loop cover message. // It kinda seems like there must exist "prefix" or "starts_with" method for bytes // or something, but I couldn't find anything diff --git a/gateway/src/node/mixnet_handling/receiver/connection_handler.rs b/gateway/src/node/mixnet_handling/receiver/connection_handler.rs index d07dc6489e..58f60ef581 100644 --- a/gateway/src/node/mixnet_handling/receiver/connection_handler.rs +++ b/gateway/src/node/mixnet_handling/receiver/connection_handler.rs @@ -155,7 +155,15 @@ impl ConnectionHandler { let client_address = processed_final_hop.destination; let message = processed_final_hop.message; let forward_ack = processed_final_hop.forward_ack; - + if client_address + == DestinationAddressBytes::from_bytes([ + 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, + 0, 0, 0, 0, + ]) + { + trace!("NULL address, dropping the packet"); + return Ok(()); + } // we failed to push message directly to the client - it's probably offline. // we should store it on the disk instead. match self.try_push_message_to_client(client_address, message) {