diff --git a/clients/webassembly/src/client/mod.rs b/clients/webassembly/src/client/mod.rs index 44f257a6e6..9d320469e3 100644 --- a/clients/webassembly/src/client/mod.rs +++ b/clients/webassembly/src/client/mod.rs @@ -31,7 +31,6 @@ use nymsphinx::acknowledgements::AckKey; use nymsphinx::addressing::clients::Recipient; use nymsphinx::preparer::MessagePreparer; use rand::rngs::OsRng; -use received_processor::ReceivedMessagesProcessor; use std::sync::Arc; use std::time::Duration; use topology::{gateway, nym_topology_from_bonds, NymTopology}; @@ -41,7 +40,6 @@ use wasm_bindgen_futures::spawn_local; use wasm_utils::{console_log, console_warn}; pub mod config; -pub(crate) mod received_processor; // TODO: make those properly configurable later const ACK_WAIT_MULTIPLIER: f64 = 1.5; diff --git a/clients/webassembly/src/client/received_processor.rs b/clients/webassembly/src/client/received_processor.rs deleted file mode 100644 index a59a0dfa59..0000000000 --- a/clients/webassembly/src/client/received_processor.rs +++ /dev/null @@ -1,183 +0,0 @@ -// Copyright 2021 - Nym Technologies SA -// SPDX-License-Identifier: Apache-2.0 - -use crypto::asymmetric::encryption; -use futures::StreamExt; -use gateway_client::{AcknowledgementReceiver, MixnetMessageReceiver}; -use nymsphinx::acknowledgements::identifier::recover_identifier; -use nymsphinx::acknowledgements::AckKey; -use nymsphinx::chunking::fragment::{FragmentIdentifier, COVER_FRAG_ID}; -use nymsphinx::receiver::{MessageReceiver, MessageRecoveryError, ReconstructedMessage}; -use serde::{Deserialize, Serialize}; -use std::collections::HashSet; -use std::sync::Arc; -use wasm_bindgen::JsValue; -use wasm_utils::{console_error, console_log, console_warn}; - -#[derive(Serialize, Deserialize)] -pub struct ProcessedMessage { - pub message: String, - pub reply_surb: Option, -} - -impl From for ProcessedMessage { - fn from(reconstructed: ReconstructedMessage) -> Self { - ProcessedMessage { - message: String::from_utf8_lossy(&reconstructed.message).into_owned(), - reply_surb: reconstructed - .reply_surb - .map(|reply_surb| reply_surb.to_base58_string()), - } - } -} - -pub(crate) struct ReceivedMessagesProcessor { - local_encryption_keypair: Arc, - ack_key: Arc, - message_receiver: MessageReceiver, - - recently_reconstructed: HashSet, -} - -impl ReceivedMessagesProcessor { - pub(crate) fn new( - local_encryption_keypair: Arc, - ack_key: Arc, - ) -> Self { - ReceivedMessagesProcessor { - local_encryption_keypair, - ack_key, - message_receiver: MessageReceiver::new(), - recently_reconstructed: HashSet::new(), - } - } - - // TODO: duplicate code from received_buffer.rs in client-core.... - fn process_received_fragment(&mut self, raw_fragment: Vec) -> Option { - let fragment_data = match self - .message_receiver - .recover_plaintext(self.local_encryption_keypair.private_key(), raw_fragment) - { - Err(e) => { - console_warn!("failed to recover fragment data: {:?}. The whole underlying message might be corrupted and unrecoverable!", e); - return None; - } - Ok(frag_data) => frag_data, - }; - - if nymsphinx::cover::is_cover(&fragment_data) { - // currently won't be the case for a loooong time - console_log!("The message was a loop cover message! Skipping it"); - return None; - } - - let fragment = match self.message_receiver.recover_fragment(&fragment_data) { - Err(e) => { - console_warn!("failed to recover fragment from raw data: {:?}. The whole underlying message might be corrupted and unrecoverable!", e); - return None; - } - Ok(frag) => frag, - }; - - if self.recently_reconstructed.contains(&fragment.id()) { - console_warn!("Received a chunk of already re-assembled message ({:?})! It probably got here because the ack got lost", fragment.id()); - return None; - } - - // if we returned an error the underlying message is malformed in some way - match self.message_receiver.insert_new_fragment(fragment) { - Err(err) => match err { - MessageRecoveryError::MalformedReconstructedMessage(message_sets) => { - // TODO: should we really insert reconstructed sets? could this be abused for some attack? - for set_id in message_sets { - if !self.recently_reconstructed.insert(set_id) { - // or perhaps we should even panic at this point? - console_error!( - "Reconstructed another message containing already used set id!" - ) - } - } - None - } - _ => unreachable!( - "no other error kind should have been returned here! If so, it's a bug!" - ), - }, - Ok(reconstruction_result) => match reconstruction_result { - Some((reconstructed_message, used_sets)) => { - for set_id in used_sets { - if !self.recently_reconstructed.insert(set_id) { - // or perhaps we should even panic at this point? - console_error!( - "Reconstructed another message containing already used set id!" - ) - } - } - Some(reconstructed_message.into()) - } - None => None, - }, - } - } - - // TODO: duplicate code from acknowledgement listener... - fn process_received_ack(&self, ack_content: Vec) { - let frag_id = match recover_identifier(&self.ack_key, &ack_content) - .map(FragmentIdentifier::try_from_bytes) - { - Some(Ok(frag_id)) => frag_id, - _ => { - console_warn!("Received invalid ACK!"); // should we do anything else about that? - return; - } - }; - - // if we received an ack for cover message or a reply there will be nothing to remove, - // because nothing was inserted in the first place - if frag_id == COVER_FRAG_ID { - return; - } else if frag_id.is_reply() { - console_warn!("Received an ack for a reply message - no need to do anything! (don't know what to do!)"); - // TODO: probably there will need to be some extra procedure here, something to notify - // user that his reply reached the recipient (since we got an ack) - return; - } - - console_log!( - "Received an ack for fragment {:?} but can't do anything more about it just yet...", - frag_id - ); - - // here be ack handling... - } - - // TODO: this needs to have a shutdown signal! - pub(crate) async fn start_processing( - mut self, - mixnet_messages_receiver: MixnetMessageReceiver, - ack_receiver: AcknowledgementReceiver, - on_message: js_sys::Function, - ) { - let mut fused_mixnet_messages_receiver = mixnet_messages_receiver.fuse(); - let mut fused_ack_receiver = ack_receiver.fuse(); - let this = JsValue::null(); - - loop { - futures::select! { - mix_msgs = fused_mixnet_messages_receiver.next() => { - for mix_msg in mix_msgs.unwrap() { - if let Some(processed) = self.process_received_fragment(mix_msg) { - let arg1 = JsValue::from_serde(&processed).unwrap(); - on_message.call1(&this, &arg1).expect("on message failed!"); - } - } - } - acks = fused_ack_receiver.next() => { - for ack in acks.unwrap() { - self.process_received_ack(ack); - } - } - } - } - } -}