diff --git a/common/addressing/src/lib.rs b/common/addressing/src/lib.rs index a2660471f4..529d93924d 100644 --- a/common/addressing/src/lib.rs +++ b/common/addressing/src/lib.rs @@ -62,8 +62,8 @@ pub fn encoded_bytes_from_socket_address(address: SocketAddr) -> [u8; 32] { address_bytes } -pub fn socket_address_from_encoded_bytes(b: [u8; 32]) -> SocketAddr { - let address_type: AddressType = b[0].try_into().unwrap(); +pub fn socket_address_from_encoded_bytes(b: [u8; 32]) -> Result { + let address_type: AddressType = b[0].try_into()?; let port: u16 = u16::from_be_bytes([b[1], b[2]]); @@ -76,5 +76,5 @@ pub fn socket_address_from_encoded_bytes(b: [u8; 32]) -> SocketAddr { } }; - SocketAddr::new(ip, port) + Ok(SocketAddr::new(ip, port)) } diff --git a/common/clients/directory-client/src/presence.rs b/common/clients/directory-client/src/presence.rs index f56305f172..7d1b30c24b 100644 --- a/common/clients/directory-client/src/presence.rs +++ b/common/clients/directory-client/src/presence.rs @@ -9,6 +9,7 @@ use std::net::ToSocketAddrs; use topology::coco; use topology::mix; use topology::provider; +use topology::provider::Node; use topology::NymTopology; // special version of 'PartialEq' that does not care about last_seen field (where applicable) @@ -62,16 +63,18 @@ impl TryInto for MixNodePresence { type Error = io::Error; fn try_into(self) -> Result { - let resolved_hostname = self.host.to_socket_addrs()?.next(); - if resolved_hostname.is_none() { - return Err(io::Error::new( - io::ErrorKind::Other, - "no valid socket address", - )); - } + let resolved_hostname = match self.host.to_socket_addrs()?.next() { + None => { + return Err(io::Error::new( + io::ErrorKind::Other, + "no valid socket address", + )); + } + Some(host) => host, + }; Ok(topology::mix::Node { - host: resolved_hostname.unwrap(), + host: resolved_hostname, pub_key: self.pub_key, layer: self.layer, last_seen: self.last_seen, @@ -129,11 +132,13 @@ impl PresenceEq for MixProviderPresence { } } -impl Into for MixProviderPresence { - fn into(self) -> topology::provider::Node { - topology::provider::Node { - client_listener: self.client_listener.parse().unwrap(), - mixnet_listener: self.mixnet_listener.parse().unwrap(), +impl TryInto for MixProviderPresence { + type Error = std::net::AddrParseError; + + fn try_into(self) -> Result { + Ok(topology::provider::Node { + client_listener: self.client_listener.parse()?, + mixnet_listener: self.mixnet_listener.parse()?, pub_key: self.pub_key, registered_clients: self .registered_clients @@ -142,7 +147,7 @@ impl Into for MixProviderPresence { .collect(), last_seen: self.last_seen, version: self.version, - } + }) } } @@ -345,7 +350,7 @@ impl NymTopology for Topology { fn providers(&self) -> Vec { self.mix_provider_nodes .iter() - .map(|x| x.clone().into()) + .filter_map(|x| x.clone().try_into().ok()) .collect() } diff --git a/common/clients/mix-client/src/packet.rs b/common/clients/mix-client/src/packet.rs index 7999d1e620..f1ee832651 100644 --- a/common/clients/mix-client/src/packet.rs +++ b/common/clients/mix-client/src/packet.rs @@ -1,17 +1,49 @@ use addressing; +use addressing::AddressTypeError; use sphinx::route::{Destination, DestinationAddressBytes, SURBIdentifier}; use sphinx::SphinxPacket; use std::net::SocketAddr; -use topology::NymTopology; +use topology::{NymTopology, NymTopologyError}; pub const LOOP_COVER_MESSAGE_PAYLOAD: &[u8] = b"The cake is a lie!"; pub const LOOP_COVER_MESSAGE_AVERAGE_DELAY: f64 = 2.0; +#[derive(Debug)] +pub enum SphinxPacketEncapsulationError { + NoValidProvidersError, + InvalidTopologyError, + SphinxEncapsulationError(sphinx::header::SphinxUnwrapError), + InvalidFirstMixAddress, +} + +impl From for SphinxPacketEncapsulationError { + fn from(_: NymTopologyError) -> Self { + use SphinxPacketEncapsulationError::*; + InvalidTopologyError + } +} + +// it is correct error we're converting from, it just has an unfortunate name +// related issue: https://github.com/nymtech/sphinx/issues/40 +impl From for SphinxPacketEncapsulationError { + fn from(err: sphinx::header::SphinxUnwrapError) -> Self { + use SphinxPacketEncapsulationError::*; + SphinxEncapsulationError(err) + } +} + +impl From for SphinxPacketEncapsulationError { + fn from(_: AddressTypeError) -> Self { + use SphinxPacketEncapsulationError::*; + InvalidFirstMixAddress + } +} + pub fn loop_cover_message( our_address: DestinationAddressBytes, surb_id: SURBIdentifier, topology: &T, -) -> (SocketAddr, SphinxPacket) { +) -> Result<(SocketAddr, SphinxPacket), SphinxPacketEncapsulationError> { let destination = Destination::new(our_address, surb_id); encapsulate_message( @@ -27,19 +59,24 @@ pub fn encapsulate_message( message: Vec, topology: &T, average_delay: f64, -) -> (SocketAddr, SphinxPacket) { +) -> Result<(SocketAddr, SphinxPacket), SphinxPacketEncapsulationError> { let mut providers = topology.providers(); + if providers.len() == 0 { + return Err(SphinxPacketEncapsulationError::NoValidProvidersError); + } + // unwrap is fine here as we asserted there is at least single provider let provider = providers.pop().unwrap().into(); - let route = topology.route_to(provider).unwrap(); + let route = topology.route_to(provider)?; let delays = sphinx::header::delays::generate(route.len(), average_delay); // build the packet - let packet = sphinx::SphinxPacket::new(message, &route[..], &recipient, &delays).unwrap(); + let packet = sphinx::SphinxPacket::new(message, &route[..], &recipient, &delays)?; + // we know the mix route must be valid otherwise we would have already returned an error let first_node_address = - addressing::socket_address_from_encoded_bytes(route.first().unwrap().address.to_bytes()); + addressing::socket_address_from_encoded_bytes(route.first().unwrap().address.to_bytes())?; - (first_node_address, packet) + Ok((first_node_address, packet)) } diff --git a/common/clients/mix-client/src/poisson.rs b/common/clients/mix-client/src/poisson.rs index d4991f0dd8..6474d3c65b 100644 --- a/common/clients/mix-client/src/poisson.rs +++ b/common/clients/mix-client/src/poisson.rs @@ -1,6 +1,9 @@ use rand_distr::{Distribution, Exp}; pub fn sample(average_delay: f64) -> f64 { + // this is our internal code used by our traffic streams + // the error is only thrown if average delay is less than 0, which will never happen + // so call to unwrap is perfectly safe here let exp = Exp::new(1.0 / average_delay).unwrap(); exp.sample(&mut rand::thread_rng()) } diff --git a/common/clients/provider-client/src/lib.rs b/common/clients/provider-client/src/lib.rs index befac332c9..07e924f5af 100644 --- a/common/clients/provider-client/src/lib.rs +++ b/common/clients/provider-client/src/lib.rs @@ -66,7 +66,7 @@ impl ProviderClient { pub async fn send_request(&self, bytes: Vec) -> Result, ProviderClientError> { let mut socket = tokio::net::TcpStream::connect(self.provider_network_address).await?; - socket.set_keepalive(Some(Duration::from_secs(2))).unwrap(); + socket.set_keepalive(Some(Duration::from_secs(2)))?; socket.write_all(&bytes[..]).await?; if let Err(e) = socket.shutdown(Shutdown::Write) { warn!("failed to close write part of the socket; err = {:?}", e) @@ -82,11 +82,14 @@ impl ProviderClient { } pub async fn retrieve_messages(&self) -> Result>, ProviderClientError> { - if self.auth_token.is_none() { - return Err(ProviderClientError::EmptyAuthTokenError); - } + let auth_token = match self.auth_token { + Some(token) => token, + None => { + return Err(ProviderClientError::EmptyAuthTokenError); + } + }; - let pull_request = PullRequest::new(self.our_address, self.auth_token.unwrap()); + let pull_request = PullRequest::new(self.our_address, auth_token); let bytes = pull_request.to_bytes(); let response = self.send_request(bytes).await?; diff --git a/common/crypto/src/encryption/x25519.rs b/common/crypto/src/encryption/x25519.rs index d7d5be9f62..44d86783da 100644 --- a/common/crypto/src/encryption/x25519.rs +++ b/common/crypto/src/encryption/x25519.rs @@ -56,6 +56,7 @@ impl MixnetEncryptionPrivateKey for PrivateKey { fn from_bytes(b: &[u8]) -> Self { let mut bytes = [0; 32]; bytes.copy_from_slice(&b[..]); + // due to trait restriction we have no choice but to panic if this fails let key = Scalar::from_canonical_bytes(bytes).unwrap(); Self(key) } diff --git a/common/healthcheck/src/path_check.rs b/common/healthcheck/src/path_check.rs index f2d8b4879b..e98666485e 100644 --- a/common/healthcheck/src/path_check.rs +++ b/common/healthcheck/src/path_check.rs @@ -153,16 +153,23 @@ impl PathChecker { let mut provider_messages = Vec::new(); for provider_client in self.provider_clients.values() { // if it was none all associated paths were already marked as unhealthy - if provider_client.is_some() { - let pc = provider_client.as_ref().unwrap(); - provider_messages.extend(self.resolve_pending_provider_checks(pc).await); - } + let pc = match provider_client { + Some(pc) => pc, + None => continue, + }; + + provider_messages.extend(self.resolve_pending_provider_checks(pc).await); } self.update_path_statuses(provider_messages); } pub(crate) async fn send_test_packet(&mut self, path: &Vec, iteration: u8) { + if path.len() == 0 { + warn!("trying to send test packet through an empty path!"); + return; + } + debug!("Checking path: {:?} ({})", path, iteration); let path_identifier = PathChecker::unique_path_key(path, iteration); @@ -171,7 +178,13 @@ impl PathChecker { // does provider exist? let provider_client = self .provider_clients - .get(&path.last().unwrap().pub_key.to_bytes()) + .get( + &path + .last() + .expect("We checked the path to contain at least one entry") + .pub_key + .to_bytes(), + ) .unwrap(); if provider_client.is_none() { @@ -186,10 +199,15 @@ impl PathChecker { return; } - let layer_one_mix = path.first().unwrap(); + let layer_one_mix = path + .first() + .expect("We checked the path to contain at least one entry"); let first_node_key = layer_one_mix.pub_key.to_bytes(); + + // we generated the bytes data so unwrap is fine let first_node_address = - addressing::socket_address_from_encoded_bytes(layer_one_mix.address.to_bytes()); + addressing::socket_address_from_encoded_bytes(layer_one_mix.address.to_bytes()) + .unwrap(); let first_node_client = self .layer_one_clients @@ -208,10 +226,12 @@ impl PathChecker { return; } + // we already checked for 'None' case let first_node_client = first_node_client.as_ref().unwrap(); let delays: Vec<_> = path.iter().map(|_| Delay::new(0)).collect(); + // all of the data used to create the packet was created by us let packet = sphinx::SphinxPacket::new( path_identifier.clone(), &path[..], diff --git a/common/healthcheck/src/result.rs b/common/healthcheck/src/result.rs index b7aae8d33c..de70873e38 100644 --- a/common/healthcheck/src/result.rs +++ b/common/healthcheck/src/result.rs @@ -14,9 +14,9 @@ pub struct HealthCheckResult(Vec); impl std::fmt::Display for HealthCheckResult { fn fmt(&self, f: &mut Formatter) -> Result<(), Error> { write!(f, "NETWORK HEALTH\n==============\n")?; - self.0 - .iter() - .for_each(|score| write!(f, "{}\n", score).unwrap()); + for score in self.0.iter() { + write!(f, "{}\n", score)? + } Ok(()) } } diff --git a/common/pemstore/src/pemstore.rs b/common/pemstore/src/pemstore.rs index d16f8574fa..9c7a914d02 100644 --- a/common/pemstore/src/pemstore.rs +++ b/common/pemstore/src/pemstore.rs @@ -1,6 +1,7 @@ use crate::pathfinder::PathFinder; use pem::{encode, parse, Pem}; use std::fs::File; +use std::io; use std::io::prelude::*; use std::path::PathBuf; @@ -24,43 +25,50 @@ impl PemStore { } } - pub fn read_identity(&self) -> IDPair + pub fn read_identity(&self) -> io::Result where IDPair: crypto::identity::MixnetIdentityKeyPair, Priv: crypto::identity::MixnetIdentityPrivateKey, Pub: crypto::identity::MixnetIdentityPublicKey, { - let private_pem = self.read_pem_file(self.private_mix_key.clone()); - let public_pem = self.read_pem_file(self.public_mix_key.clone()); + let private_pem = self.read_pem_file(self.private_mix_key.clone())?; + let public_pem = self.read_pem_file(self.public_mix_key.clone())?; let key_pair = IDPair::from_bytes(&private_pem.contents, &public_pem.contents); - assert_eq!(key_pair.private_key().pem_type(), private_pem.tag); - assert_eq!(key_pair.public_key().pem_type(), public_pem.tag); + if key_pair.private_key().pem_type() != private_pem.tag { + return Err(io::Error::new( + io::ErrorKind::Other, + "unexpected private key pem tag", + )); + } - key_pair + if key_pair.public_key().pem_type() != public_pem.tag { + return Err(io::Error::new( + io::ErrorKind::Other, + "unexpected public key pem tag", + )); + } + + Ok(key_pair) } - fn read_pem_file(&self, filepath: PathBuf) -> Pem { - let mut pem_bytes = File::open(filepath).expect("Could not open stored keys from disk."); + fn read_pem_file(&self, filepath: PathBuf) -> io::Result { + let mut pem_bytes = File::open(filepath)?; let mut buf = Vec::new(); - pem_bytes - .read_to_end(&mut buf) - .expect("PEM bytes reading failed."); - let pem = parse(&buf).expect("PEM parsing failed while reading keys"); - - pem + pem_bytes.read_to_end(&mut buf)?; + parse(&buf).map_err(|e| io::Error::new(io::ErrorKind::Other, e)) } // This should be refactored and made more generic for when we have other kinds of // KeyPairs that we want to persist (e.g. validator keypairs, or keys for // signing vs encryption). However, for the moment, it does the job. - pub fn write_identity(&self, key_pair: IDPair) + pub fn write_identity(&self, key_pair: IDPair) -> io::Result<()> where IDPair: crypto::identity::MixnetIdentityKeyPair, Priv: crypto::identity::MixnetIdentityPrivateKey, Pub: crypto::identity::MixnetIdentityPublicKey, { - std::fs::create_dir_all(self.config_dir.clone()).unwrap(); + std::fs::create_dir_all(self.config_dir.clone())?; let private_key = key_pair.private_key(); let public_key = key_pair.public_key(); @@ -68,22 +76,25 @@ impl PemStore { self.private_mix_key.clone(), private_key.to_bytes(), private_key.pem_type(), - ); + )?; self.write_pem_file( self.public_mix_key.clone(), public_key.to_bytes(), public_key.pem_type(), - ); + )?; + Ok(()) } - fn write_pem_file(&self, filepath: PathBuf, data: Vec, tag: String) { + fn write_pem_file(&self, filepath: PathBuf, data: Vec, tag: String) -> io::Result<()> { let pem = Pem { tag, contents: data, }; let key = encode(&pem); - let mut file = File::create(filepath).unwrap(); - file.write_all(key.as_bytes()).unwrap(); + let mut file = File::create(filepath)?; + file.write_all(key.as_bytes())?; + + Ok(()) } } diff --git a/common/topology/src/lib.rs b/common/topology/src/lib.rs index 94ac5a74b4..1004ef7ca3 100644 --- a/common/topology/src/lib.rs +++ b/common/topology/src/lib.rs @@ -57,6 +57,8 @@ pub trait NymTopology: Sized + PartialEq + std::fmt::Debug + Send + Sync { let mut layered_topology = self.make_layered_topology()?; let num_layers = layered_topology.len(); let route = (1..=num_layers as u64) + // unwrap is safe for 'remove' as it it failed, it implied the entry never existed + // in the map in the first place which would contradict what we've just done .map(|layer| layered_topology.remove(&layer).unwrap()) // for each layer .map(|nodes| nodes.into_iter().choose(&mut rand::thread_rng()).unwrap()) // choose random node .map(|random_node| random_node.into()) // and convert it into sphinx specific node format diff --git a/mixnode/src/mix_peer.rs b/mixnode/src/mix_peer.rs index 1ddb0a3481..8a8579a780 100644 --- a/mixnode/src/mix_peer.rs +++ b/mixnode/src/mix_peer.rs @@ -1,4 +1,5 @@ use addressing; +use addressing::AddressTypeError; use sphinx::route::NodeAddressBytes; use std::error::Error; use std::net::SocketAddr; @@ -9,14 +10,27 @@ pub struct MixPeer { connection: SocketAddr, } +#[derive(Debug)] +pub enum MixPeerError { + InvalidAddressError, +} + +impl From for MixPeerError { + fn from(_: AddressTypeError) -> Self { + use MixPeerError::*; + + InvalidAddressError + } +} + impl MixPeer { // note that very soon `next_hop_address` will be changed to `next_hop_metadata` - pub fn new(next_hop_address: NodeAddressBytes) -> MixPeer { + pub fn new(next_hop_address: NodeAddressBytes) -> Result { let next_hop_socket_address = - addressing::socket_address_from_encoded_bytes(next_hop_address.to_bytes()); - MixPeer { + addressing::socket_address_from_encoded_bytes(next_hop_address.to_bytes())?; + Ok(MixPeer { connection: next_hop_socket_address, - } + }) } pub async fn send(&self, bytes: Vec) -> Result<(), Box> { diff --git a/mixnode/src/node/mod.rs b/mixnode/src/node/mod.rs index c8ce9ae783..4ac744ce76 100644 --- a/mixnode/src/node/mod.rs +++ b/mixnode/src/node/mod.rs @@ -40,6 +40,7 @@ pub enum MixProcessingError { SphinxRecoveryError, ReceivedFinalHopError, SphinxProcessingError, + InvalidHopAddress, } impl From for MixProcessingError { @@ -109,9 +110,14 @@ impl PacketProcessor { ) -> Result { // we received something resembling a sphinx packet, report it! let processing_data = processing_data.lock().await; - let mut received_sender = processing_data.received_metrics_tx.clone(); + let mut received_metrics_tx = processing_data.received_metrics_tx.clone(); - received_sender.send(()).await.unwrap(); + // if unwrap failed it means our metrics reporter died, so we should exit application and + // force restart + if received_metrics_tx.send(()).await.is_err() { + error!("failed to send metrics data to the controller - the underlying thread probably died!"); + std::process::exit(1); + } let packet = SphinxPacket::from_bytes(packet_data.to_vec())?; let (next_packet, next_hop_address, delay) = @@ -126,7 +132,10 @@ impl PacketProcessor { } }; - let next_mix = MixPeer::new(next_hop_address); + 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, @@ -140,11 +149,16 @@ impl PacketProcessor { 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; - forwarding_data + + if forwarding_data .sent_metrics_tx .send(forwarding_data.recipient.to_string()) .await - .unwrap(); + .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 @@ -188,7 +202,6 @@ impl MixNode { mut socket: tokio::net::TcpStream, processing_data: Arc>, ) { - // NOTE: processing_data is copied here!! let mut buf = [0u8; sphinx::PACKET_SIZE]; // In a loop, read data from the socket and write the data back. @@ -200,12 +213,18 @@ impl MixNode { return; } Ok(_) => { - let fwd_data = PacketProcessor::process_sphinx_data_packet( + let fwd_data = match PacketProcessor::process_sphinx_data_packet( buf.as_ref(), processing_data.clone(), ) .await - .unwrap(); + { + Ok(fwd_data) => fwd_data, + Err(e) => { + warn!("failed to process sphinx packet: {:?}", e); + return; + } + }; PacketProcessor::wait_and_forward(fwd_data).await; } Err(e) => { diff --git a/nym-client/src/client/cover_traffic_stream.rs b/nym-client/src/client/cover_traffic_stream.rs index f1caedc740..c48c0c4801 100644 --- a/nym-client/src/client/cover_traffic_stream.rs +++ b/nym-client/src/client/cover_traffic_stream.rs @@ -2,7 +2,7 @@ use crate::client::mix_traffic::MixMessage; use crate::client::topology_control::TopologyInnerRef; use crate::client::LOOP_COVER_AVERAGE_DELAY; use futures::channel::mpsc; -use log::{info, trace, warn}; +use log::{error, info, trace, warn}; use sphinx::route::Destination; use std::time::Duration; use topology::NymTopology; @@ -20,16 +20,28 @@ pub(crate) async fn start_loop_cover_traffic_stream( tokio::time::delay_for(delay_duration).await; let read_lock = topology_ctrl_ref.read().await; - let topology = read_lock.topology.as_ref(); - if topology.is_none() { - warn!("No valid topology detected - won't send any loop cover message this time"); - continue; - } + let topology = match read_lock.topology.as_ref() { + None => { + warn!("No valid topology detected - won't send any loop cover message this time"); + continue; + } + Some(topology) => topology, + }; - let topology = topology.unwrap(); - - let cover_message = - mix_client::packet::loop_cover_message(our_info.address, our_info.identifier, topology); + let cover_message = match mix_client::packet::loop_cover_message( + our_info.address, + our_info.identifier, + topology, + ) { + Ok(message) => message, + Err(err) => { + error!( + "Somehow we managed to create an invalid cover message - {:?}", + err + ); + continue; + } + }; // if this one fails, there's no retrying because it means that either: // - we run out of memory diff --git a/nym-client/src/client/real_traffic_stream.rs b/nym-client/src/client/real_traffic_stream.rs index 9d1ee055ab..fc8069c12b 100644 --- a/nym-client/src/client/real_traffic_stream.rs +++ b/nym-client/src/client/real_traffic_stream.rs @@ -4,7 +4,7 @@ use crate::client::{InputMessage, MESSAGE_SENDING_AVERAGE_DELAY}; use futures::channel::mpsc; use futures::task::{Context, Poll}; use futures::{Future, Stream, StreamExt}; -use log::{info, trace, warn}; +use log::{error, info, trace, warn}; use sphinx::route::Destination; use std::pin::Pin; use std::time::Duration; @@ -85,14 +85,15 @@ impl OutQueueControl { while let Some(next_message) = self.next().await { trace!("created new message"); let read_lock = self.topology_ctrl_ref.read().await; - let topology = read_lock.topology.as_ref(); - - if topology.is_none() { - warn!("No valid topology detected - won't send any loop cover or real message this time"); - continue; - } - - let topology = topology.unwrap(); + let topology = match read_lock.topology.as_ref() { + None => { + warn!( + "No valid topology detected - won't send any loop cover message this time" + ); + continue; + } + Some(topology) => topology, + }; let next_packet = match next_message { StreamMessage::Cover => mix_client::packet::loop_cover_message( @@ -108,6 +109,17 @@ impl OutQueueControl { ), }; + let next_packet = match next_packet { + Ok(message) => message, + Err(err) => { + error!( + "Somehow we managed to create an invalid traffic message - {:?}", + err + ); + continue; + } + }; + // if this one fails, there's no retrying because it means that either: // - we run out of memory // - the receiver channel is closed diff --git a/nym-client/src/commands/init.rs b/nym-client/src/commands/init.rs index 4ac9c47db4..bc4403e6b9 100644 --- a/nym-client/src/commands/init.rs +++ b/nym-client/src/commands/init.rs @@ -12,7 +12,7 @@ pub fn execute(matches: &ArgMatches) { println!("Writing keypairs to {:?}...", pathfinder.config_dir); let mix_keys = crypto::identity::DummyMixIdentityKeyPair::new(); let pem_store = PemStore::new(pathfinder); - pem_store.write_identity(mix_keys); + pem_store.write_identity(mix_keys).unwrap(); println!("Client configuration completed.\n\n\n") } diff --git a/nym-client/src/commands/tcpsocket.rs b/nym-client/src/commands/tcpsocket.rs index 81e0ef72d5..87901f5543 100644 --- a/nym-client/src/commands/tcpsocket.rs +++ b/nym-client/src/commands/tcpsocket.rs @@ -27,7 +27,9 @@ pub fn execute(matches: &ArgMatches) { .expect("Failed to extract the socket address from the iterator"); // TODO: currently we know we are reading the 'DummyMixIdentityKeyPair', but how to properly assert the type? - let keypair: DummyMixIdentityKeyPair = PemStore::new(ClientPathfinder::new(id)).read_identity(); + let keypair: DummyMixIdentityKeyPair = PemStore::new(ClientPathfinder::new(id)) + .read_identity() + .unwrap(); // TODO: reading auth_token from disk (if exists); println!("Public key: {}", keypair.public_key.to_b64_string()); diff --git a/nym-client/src/commands/websocket.rs b/nym-client/src/commands/websocket.rs index 644e023306..5359c10871 100644 --- a/nym-client/src/commands/websocket.rs +++ b/nym-client/src/commands/websocket.rs @@ -27,7 +27,9 @@ pub fn execute(matches: &ArgMatches) { .expect("Failed to extract the socket address from the iterator"); // TODO: currently we know we are reading the 'DummyMixIdentityKeyPair', but how to properly assert the type? - let keypair: DummyMixIdentityKeyPair = PemStore::new(ClientPathfinder::new(id)).read_identity(); + let keypair: DummyMixIdentityKeyPair = PemStore::new(ClientPathfinder::new(id)) + .read_identity() + .unwrap(); // TODO: reading auth_token from disk (if exists); diff --git a/nym-client/src/sockets/tcp.rs b/nym-client/src/sockets/tcp.rs index 8f49707c60..69440809d3 100644 --- a/nym-client/src/sockets/tcp.rs +++ b/nym-client/src/sockets/tcp.rs @@ -120,15 +120,16 @@ impl ClientRequest { return ServerResponse::Error { message: e }; } - let messages = res_rx.map(|msg| msg).await; + let messages = match res_rx.map(|msg| msg).await { + Ok(messages) => messages, + Err(e) => { + warn!("Failed to fetch client messages - {:?}", e); + return ServerResponse::Error { + message: format!("Server failed to receive messages").to_string(), + }; + } + }; - if messages.is_err() { - return ServerResponse::Error { - message: "Server failed to receive messages".to_string(), - }; - } - - let messages = messages.unwrap(); trace!("fetched {} messages", messages.len()); ServerResponse::Fetch { messages } } diff --git a/nym-client/src/sockets/ws.rs b/nym-client/src/sockets/ws.rs index 28d8b0e196..8840e6f95e 100644 --- a/nym-client/src/sockets/ws.rs +++ b/nym-client/src/sockets/ws.rs @@ -234,16 +234,17 @@ impl ClientRequest { }; } - let messages = res_rx.map(|msg| msg).await; - if messages.is_err() { - warn!("Failed to handle_fetch. messages is an error"); - return ServerResponse::Error { - message: "Server failed to receive messages".to_string(), - }; - } + let messages = match res_rx.map(|msg| msg).await { + Ok(messages) => messages, + Err(e) => { + warn!("Failed to fetch client messages - {:?}", e); + return ServerResponse::Error { + message: format!("Server failed to receive messages").to_string(), + }; + } + }; let messages = messages - .unwrap() .into_iter() .map(|message| { match std::str::from_utf8(&message) { diff --git a/sfw-provider/sfw-provider-requests/src/responses.rs b/sfw-provider/sfw-provider-requests/src/responses.rs index 870f527334..c15017c76d 100644 --- a/sfw-provider/sfw-provider-requests/src/responses.rs +++ b/sfw-provider/sfw-provider-requests/src/responses.rs @@ -45,8 +45,8 @@ fn read_be_u16(input: &mut &[u8]) -> u16 { u16::from_be_bytes(int_bytes.try_into().unwrap()) } -// TODO: currently this allows for maximum 64kB payload - if we go over that in sphinx, -// we need to update this code. +// TODO: currently this allows for maximum 64kB payload - +// if we go over that in sphinx we need to update this code. impl ProviderResponse for PullResponse { // num_msgs || len1 || len2 || ... || msg1 || msg2 || ... fn to_bytes(&self) -> Vec { diff --git a/sfw-provider/src/provider/client_handling/mod.rs b/sfw-provider/src/provider/client_handling/mod.rs index b79c1962f7..034e946152 100644 --- a/sfw-provider/src/provider/client_handling/mod.rs +++ b/sfw-provider/src/provider/client_handling/mod.rs @@ -154,8 +154,9 @@ impl ClientRequestProcessor { } fn generate_new_auth_token(data: Vec, key: DummyMixIdentityPrivateKey) -> AuthToken { - let mut auth_token_raw = - HmacSha256::new_varkey(&key.to_bytes()).expect("HMAC can take key of any size"); + // also note that `new_varkey` doesn't even have an execution branch returning an error + let mut auth_token_raw = HmacSha256::new_varkey(&key.to_bytes()) + .expect("HMAC should be able take key of any size"); auth_token_raw.input(&data); let mut auth_token = [0u8; 32]; auth_token.copy_from_slice(&auth_token_raw.result().code().to_vec()); diff --git a/sfw-provider/src/provider/mix_handling/mod.rs b/sfw-provider/src/provider/mix_handling/mod.rs index c7a4e8cc02..338943efd1 100644 --- a/sfw-provider/src/provider/mix_handling/mod.rs +++ b/sfw-provider/src/provider/mix_handling/mod.rs @@ -1,6 +1,6 @@ use crate::provider::storage::StoreData; use crypto::identity::DummyMixIdentityPrivateKey; -use log::warn; +use log::{error, warn}; use sphinx::{ProcessedPacket, SphinxPacket}; use std::path::PathBuf; use std::sync::{Arc, RwLock}; @@ -62,7 +62,13 @@ impl MixPacketProcessor { processing_data: &RwLock, ) -> Result { let packet = SphinxPacket::from_bytes(packet_data.to_vec())?; - let read_processing_data = processing_data.read().unwrap(); + let read_processing_data = match processing_data.read() { + Ok(guard) => guard, + Err(e) => { + error!("processing data lock was poisoned! - {:?}", e); + std::process::exit(1) + } + }; let (client_address, client_surb_id, payload) = match packet.process(read_processing_data.secret_key.as_scalar()) { Ok(ProcessedPacket::ProcessedPacketFinalHop(client_address, surb_id, payload)) => { diff --git a/sfw-provider/src/provider/mod.rs b/sfw-provider/src/provider/mod.rs index bc32073da8..9e1f766f48 100644 --- a/sfw-provider/src/provider/mod.rs +++ b/sfw-provider/src/provider/mod.rs @@ -147,9 +147,16 @@ impl ServiceProvider { return; } }; + let processing_data_lock = match processing_data.read() { + Ok(guard) => guard, + Err(e) => { + error!("processing data lock was poisoned! - {:?}", e); + std::process::exit(1) + } + }; ClientStorage::store_processed_data( store_data, - processing_data.read().unwrap().store_dir.as_path(), + processing_data_lock.store_dir.as_path(), ) .unwrap_or_else(|e| { error!("failed to store processed sphinx message; err = {:?}", e); diff --git a/sfw-provider/src/provider/storage/mod.rs b/sfw-provider/src/provider/storage/mod.rs index fb7607a506..d19d427b55 100644 --- a/sfw-provider/src/provider/storage/mod.rs +++ b/sfw-provider/src/provider/storage/mod.rs @@ -90,18 +90,10 @@ impl ClientStorage { } let msgs: Vec<_> = std::fs::read_dir(full_store_dir)? - .map(|entry| entry.unwrap()) - .filter(|entry| { - let is_file = entry.metadata().unwrap().is_file(); - if !is_file { - error!( - "potentially corrupted client inbox! - found a non-file - {:?}", - entry.path() - ); - } - is_file - }) + .filter_map(|entry| entry.ok()) + .filter(|entry| ClientStorage::is_valid_file(entry)) .map(|entry| { + // Not yet sure how to exactly get rid of those unwraps let content = std::fs::read(entry.path()).unwrap(); ClientStorage::delete_file(entry.path()).unwrap(); content @@ -113,6 +105,30 @@ impl ClientStorage { Ok(msgs) } + fn is_valid_file(entry: &std::fs::DirEntry) -> bool { + let metadata = match entry.metadata() { + Ok(meta) => meta, + Err(e) => { + error!( + "potentially corrupted client inbox! ({:?} - failed to read its metadata - {:?}", + entry.path(), + e, + ); + return false; + } + }; + + let is_file = metadata.is_file(); + if !is_file { + error!( + "potentially corrupted client inbox! - found a non-file - {:?}", + entry.path() + ); + } + + is_file + } + // TODO: THIS NEEDS A LOCKING MECHANISM!!! (or a db layer on top - basically 'ClientStorage' on steroids) // TODO 2: This should only be called AFTER we sent the reply. Because if client's connection failed after sending request // the messages would be deleted but he wouldn't have received them