diff --git a/mixnode/src/node/packet_processing.rs b/mixnode/src/node/packet_processing.rs index 4080af472a..0a8ab92d29 100644 --- a/mixnode/src/node/packet_processing.rs +++ b/mixnode/src/node/packet_processing.rs @@ -1,4 +1,5 @@ use crate::node::metrics; +use addressing::AddressTypeError; use crypto::encryption; use log::*; use sphinx::header::delays::Delay as SphinxDelay; @@ -6,6 +7,8 @@ use sphinx::route::NodeAddressBytes; use sphinx::{ProcessedPacket, SphinxPacket}; use std::ops::Deref; use std::sync::Arc; +use std::time::Duration; +use std::net::SocketAddr; #[derive(Debug)] pub enum MixProcessingError { @@ -15,6 +18,11 @@ pub enum MixProcessingError { InvalidHopAddress, } +pub enum MixProcessingResult { + ForwardHop(SocketAddr, Vec), + LoopMessage, +} + impl From for MixProcessingError { // for time being just have a single error instance for all possible results of sphinx::ProcessingError fn from(_: sphinx::ProcessingError) -> Self { @@ -24,30 +32,14 @@ impl From for MixProcessingError { } } -// -// struct ForwardingData { -// packet: SphinxPacket, -// delay: SphinxDelay, -// recipient: MixPeer, -// sent_metrics_tx: mpsc::Sender, -// } -// -// // TODO: this will need to be changed if MixPeer will live longer than our Forwarding Data -// impl ForwardingData { -// fn new( -// packet: SphinxPacket, -// delay: SphinxDelay, -// recipient: MixPeer, -// sent_metrics_tx: mpsc::Sender, -// ) -> Self { -// ForwardingData { -// packet, -// delay, -// recipient, -// sent_metrics_tx, -// } -// } -// } +impl From for MixProcessingError { + fn from(_: AddressTypeError) -> Self { + use MixProcessingError::*; + + InvalidHopAddress + } +} + // PacketProcessor contains all data required to correctly unwrap and forward sphinx packets #[derive(Clone)] @@ -66,64 +58,33 @@ impl PacketProcessor { metrics_reporter, } } + + pub(crate) fn report_sent(&self, addr: SocketAddr) { + self.metrics_reporter.report_sent(addr.to_string()) + } - // async fn wait_and_forward(mut forwarding_data: ForwardingData) { - // let delay_duration = Duration::from_nanos(forwarding_data.delay.get_value()); - // tokio::time::delay_for(delay_duration).await; - // - // if forwarding_data - // .sent_metrics_tx - // .send(forwarding_data.recipient.stringify()) - // .await - // .is_err() - // { - // error!("failed to send metrics data to the controller - the underlying thread probably died!"); - // std::process::exit(1); - // } - // - // trace!("RECIPIENT: {:?}", forwarding_data.recipient); - // match forwarding_data - // .recipient - // .send(forwarding_data.packet.to_bytes()) - // .await - // { - // Ok(()) => (), - // Err(e) => { - // warn!( - // "failed to write bytes to next mix peer. err = {:?}", - // e.to_string() - // ); - // } - // } - // } - - fn process_forward_hop( + async fn process_forward_hop( &self, packet: SphinxPacket, forward_address: NodeAddressBytes, delay: SphinxDelay, - ) -> Result<(), MixProcessingError> { - unimplemented!() - // let next_mix = match MixPeer::new(next_hop_address) { - // Ok(next_mix) => next_mix, - // Err(_) => return Err(MixProcessingError::InvalidHopAddress), - // }; - // - // let fwd_data = ForwardingData::new( - // next_packet, - // delay, - // next_mix, - // processing_data.sent_metrics_tx.clone(), - // ); - // Ok(fwd_data) + ) -> Result { + let next_hop_address = + addressing::socket_address_from_encoded_bytes(forward_address.to_bytes())?; - // TODO: do forwarding here? + // TODO: change the calls after merge on sphinx PR + let delay_duration = Duration::from_nanos(delay.get_value()); + + // Delay packet for as long as required + tokio::time::delay_for(delay_duration).await; + + Ok(MixProcessingResult::ForwardHop(next_hop_address, packet.to_bytes())) } - pub(crate) fn process_sphinx_packet( + pub(crate) async fn process_sphinx_packet( &self, raw_packet_data: [u8; sphinx::PACKET_SIZE], - ) -> Result<(), MixProcessingError> { + ) -> Result { // we received something resembling a sphinx packet, report it! self.metrics_reporter.report_received(); @@ -131,7 +92,7 @@ impl PacketProcessor { match packet.process(self.secret_key.deref().inner()) { Ok(ProcessedPacket::ProcessedPacketForwardHop(packet, address, delay)) => { - self.process_forward_hop(packet, address, delay) + self.process_forward_hop(packet, address, delay).await } Ok(ProcessedPacket::ProcessedPacketFinalHop(_, _, _)) => { warn!("Received a loop cover message that we haven't implemented yet!"); @@ -144,3 +105,6 @@ impl PacketProcessor { } } } + +// TODO: the test that definitely needs to be written is as follows: +// we are stuck trying to write to mix A, can we still forward just fine to mix B?