Beginning of listener and packet processing refactoring
This commit is contained in:
Jedrzej Stuczynski
@@ -0,0 +1,87 @@
|
||||
use crate::node::packet_processing::PacketProcessor;
|
||||
use log::*;
|
||||
use std::io;
|
||||
use std::net::SocketAddr;
|
||||
use tokio::prelude::*;
|
||||
use tokio::runtime::Handle;
|
||||
use tokio::task::JoinHandle;
|
||||
|
||||
async fn process_received_packet(
|
||||
packet_data: [u8; sphinx::PACKET_SIZE],
|
||||
processing_data: PacketProcessor,
|
||||
) {
|
||||
// if we fail then we fail
|
||||
processing_data.process_sphinx_packet(packet_data);
|
||||
|
||||
// let fwd_data = match PacketProcessor::process_sphinx_data_packet(
|
||||
// &packet_data,
|
||||
// processing_data,
|
||||
// )
|
||||
// .await
|
||||
// {
|
||||
// Ok(fwd_data) => fwd_data,
|
||||
// Err(e) => {
|
||||
// warn!("failed to process sphinx packet: {:?}", e);
|
||||
// return;
|
||||
// }
|
||||
// };
|
||||
// PacketProcessor::wait_and_forward(fwd_data).await;
|
||||
}
|
||||
|
||||
async fn process_socket_connection(
|
||||
mut socket: tokio::net::TcpStream,
|
||||
packet_processor: PacketProcessor,
|
||||
) {
|
||||
let mut buf = [0u8; sphinx::PACKET_SIZE];
|
||||
loop {
|
||||
match socket.read(&mut buf).await {
|
||||
// socket closed
|
||||
Ok(n) if n == 0 => {
|
||||
trace!("Remote connection closed.");
|
||||
return;
|
||||
}
|
||||
Ok(n) => {
|
||||
if n != sphinx::PACKET_SIZE {
|
||||
warn!("read data of different length than expected sphinx packet size - {} (expected {})", n, sphinx::PACKET_SIZE);
|
||||
continue;
|
||||
}
|
||||
|
||||
// we must be able to handle multiple packets from same connection independently
|
||||
tokio::spawn(process_received_packet(
|
||||
buf.clone(),
|
||||
// note: processing_data is relatively cheap (and safe) to clone -
|
||||
// it contains arc to private key and metrics reporter (which is just
|
||||
// a single mpsc unbounded sender)
|
||||
// TODO: channel to tcp client
|
||||
packet_processor.clone(),
|
||||
))
|
||||
}
|
||||
Err(e) => {
|
||||
warn!(
|
||||
"failed to read from socket. Closing the connection; err = {:?}",
|
||||
e
|
||||
);
|
||||
return;
|
||||
}
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn run_socket_listener(
|
||||
handle: &Handle,
|
||||
addr: SocketAddr,
|
||||
packet_processor: PacketProcessor,
|
||||
) -> JoinHandle<io::Result<()>> {
|
||||
let handle_clone = handle.clone();
|
||||
handle.spawn(async move {
|
||||
let mut listener = tokio::net::TcpListener::bind(addr).await?;
|
||||
loop {
|
||||
let (socket, _) = listener.accept().await?;
|
||||
|
||||
let thread_packet_processor = packet_processor.clone();
|
||||
handle_clone.spawn(async move {
|
||||
process_socket_connection(socket, thread_packet_processor).await;
|
||||
});
|
||||
}
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,146 @@
|
||||
use crate::node::metrics;
|
||||
use crypto::encryption;
|
||||
use log::*;
|
||||
use sphinx::header::delays::Delay as SphinxDelay;
|
||||
use sphinx::route::NodeAddressBytes;
|
||||
use sphinx::{ProcessedPacket, SphinxPacket};
|
||||
use std::ops::Deref;
|
||||
use std::sync::Arc;
|
||||
|
||||
#[derive(Debug)]
|
||||
pub enum MixProcessingError {
|
||||
SphinxRecoveryError,
|
||||
ReceivedFinalHopError,
|
||||
SphinxProcessingError,
|
||||
InvalidHopAddress,
|
||||
}
|
||||
|
||||
impl From<sphinx::ProcessingError> for MixProcessingError {
|
||||
// for time being just have a single error instance for all possible results of sphinx::ProcessingError
|
||||
fn from(_: sphinx::ProcessingError) -> Self {
|
||||
use MixProcessingError::*;
|
||||
|
||||
SphinxRecoveryError
|
||||
}
|
||||
}
|
||||
|
||||
//
|
||||
// struct ForwardingData {
|
||||
// packet: SphinxPacket,
|
||||
// delay: SphinxDelay,
|
||||
// recipient: MixPeer,
|
||||
// sent_metrics_tx: mpsc::Sender<String>,
|
||||
// }
|
||||
//
|
||||
// // 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<String>,
|
||||
// ) -> Self {
|
||||
// ForwardingData {
|
||||
// packet,
|
||||
// delay,
|
||||
// recipient,
|
||||
// sent_metrics_tx,
|
||||
// }
|
||||
// }
|
||||
// }
|
||||
|
||||
// PacketProcessor contains all data required to correctly unwrap and forward sphinx packets
|
||||
#[derive(Clone)]
|
||||
pub struct PacketProcessor {
|
||||
secret_key: Arc<encryption::PrivateKey>,
|
||||
metrics_reporter: metrics::MetricsReporter,
|
||||
}
|
||||
|
||||
impl PacketProcessor {
|
||||
pub(crate) fn new(
|
||||
secret_key: encryption::PrivateKey,
|
||||
metrics_reporter: metrics::MetricsReporter,
|
||||
) -> Self {
|
||||
PacketProcessor {
|
||||
secret_key: Arc::new(secret_key),
|
||||
metrics_reporter,
|
||||
}
|
||||
}
|
||||
|
||||
// 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(
|
||||
&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)
|
||||
|
||||
// TODO: do forwarding here?
|
||||
}
|
||||
|
||||
pub(crate) fn process_sphinx_packet(
|
||||
&self,
|
||||
raw_packet_data: [u8; sphinx::PACKET_SIZE],
|
||||
) -> Result<(), MixProcessingError> {
|
||||
// we received something resembling a sphinx packet, report it!
|
||||
self.metrics_reporter.report_received();
|
||||
|
||||
let packet = SphinxPacket::from_bytes(&raw_packet_data)?;
|
||||
|
||||
match packet.process(self.secret_key.deref().inner()) {
|
||||
Ok(ProcessedPacket::ProcessedPacketForwardHop(packet, address, delay)) => {
|
||||
self.process_forward_hop(packet, address, delay)
|
||||
}
|
||||
Ok(ProcessedPacket::ProcessedPacketFinalHop(_, _, _)) => {
|
||||
warn!("Received a loop cover message that we haven't implemented yet!");
|
||||
Err(MixProcessingError::ReceivedFinalHopError)
|
||||
}
|
||||
Err(e) => {
|
||||
warn!("Failed to unwrap Sphinx packet: {:?}", e);
|
||||
Err(MixProcessingError::SphinxProcessingError)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user