GRIN/nym
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #101 from nymtech/feature/panic_improvements

Browse Source 
Feature/panic improvements
This commit is contained in:
Dave Hrycyszyn
2020-01-24 16:12:05 +00:00
committed by GitHub
parent 28d8fbedc9 1b787c94bf
commit ea2be1a7ec
24 changed files with 304 additions and 129 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/addressing/src/lib.rs
+3 -3
View File
@@ -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<SocketAddr, AddressTypeError> {
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))
}
common/clients/directory-client/src/presence.rs
+20 -15
View File
@@ -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<topology::mix::Node> for MixNodePresence {
type Error = io::Error;
fn try_into(self) -> Result<topology::mix::Node, Self::Error> {
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<topology::provider::Node> 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<topology::provider::Node> for MixProviderPresence {
type Error = std::net::AddrParseError;
fn try_into(self) -> Result<Node, Self::Error> {
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<topology::provider::Node> for MixProviderPresence {
.collect(),
last_seen: self.last_seen,
version: self.version,
}
})
}
}
@@ -345,7 +350,7 @@ impl NymTopology for Topology {
fn providers(&self) -> Vec<provider::Node> {
self.mix_provider_nodes
.iter()
.map(|x| x.clone().into())
.filter_map(|x| x.clone().try_into().ok())
.collect()
}
common/clients/mix-client/src/packet.rs
+44 -7
View File
@@ -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<topology::NymTopologyError> 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<sphinx::header::SphinxUnwrapError> for SphinxPacketEncapsulationError {
fn from(err: sphinx::header::SphinxUnwrapError) -> Self {
use SphinxPacketEncapsulationError::*;
SphinxEncapsulationError(err)
}
}
impl From<AddressTypeError> for SphinxPacketEncapsulationError {
fn from(_: AddressTypeError) -> Self {
use SphinxPacketEncapsulationError::*;
InvalidFirstMixAddress
}
}
pub fn loop_cover_message<T: NymTopology>(
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<T: NymTopology>(
message: Vec<u8>,
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))
}
common/clients/mix-client/src/poisson.rs
+3
View File
@@ -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())
}
common/clients/provider-client/src/lib.rs
+8 -5
View File
@@ -66,7 +66,7 @@ impl ProviderClient {
pub async fn send_request(&self, bytes: Vec<u8>) -> Result<Vec<u8>, 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<Vec<Vec<u8>>, 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?;
common/crypto/src/encryption/x25519.rs
+1
View File
@@ -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)
}
common/healthcheck/src/path_check.rs
+27 -7
View File
@@ -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<SphinxNode>, 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[..],
common/healthcheck/src/result.rs
+3 -3
View File
@@ -14,9 +14,9 @@ pub struct HealthCheckResult(Vec<NodeScore>);
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(())
}
}
common/pemstore/src/pemstore.rs
+32 -21
View File
@@ -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<IDPair, Priv, Pub>(&self) -> IDPair
pub fn read_identity<IDPair, Priv, Pub>(&self) -> io::Result<IDPair>
where
IDPair: crypto::identity::MixnetIdentityKeyPair<Priv, Pub>,
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<Pem> {
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<IDPair, Priv, Pub>(&self, key_pair: IDPair)
pub fn write_identity<IDPair, Priv, Pub>(&self, key_pair: IDPair) -> io::Result<()>
where
IDPair: crypto::identity::MixnetIdentityKeyPair<Priv, Pub>,
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<u8>, tag: String) {
fn write_pem_file(&self, filepath: PathBuf, data: Vec<u8>, 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(())
}
}
common/topology/src/lib.rs
+2
View File
@@ -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
mixnode/src/mix_peer.rs
+18 -4
View File
@@ -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<addressing::AddressTypeError> 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<MixPeer, MixPeerError> {
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<u8>) -> Result<(), Box<dyn Error>> {
mixnode/src/node/mod.rs
+27 -8
View File
@@ -40,6 +40,7 @@ pub enum MixProcessingError {
SphinxRecoveryError,
ReceivedFinalHopError,
SphinxProcessingError,
InvalidHopAddress,
}
impl From<sphinx::ProcessingError> for MixProcessingError {
@@ -109,9 +110,14 @@ impl PacketProcessor {
) -> Result<ForwardingData, MixProcessingError> {
// 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<Mutex<ProcessingData>>,
) {
// 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) => {
nym-client/src/client/cover_traffic_stream.rs
+22 -10
View File
@@ -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<T: NymTopology>(
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
nym-client/src/client/real_traffic_stream.rs
+21 -9
View File
@@ -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<T: NymTopology> OutQueueControl<T> {
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<T: NymTopology> OutQueueControl<T> {
),
};
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
nym-client/src/commands/init.rs
+1 -1
View File
@@ -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")
}
nym-client/src/commands/tcpsocket.rs
+3 -1
View File
@@ -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());
nym-client/src/commands/websocket.rs
+3 -1
View File
@@ -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);
nym-client/src/sockets/tcp.rs
+9 -8
View File
@@ -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 }
}
nym-client/src/sockets/ws.rs
+9 -8
View File
@@ -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) {
sfw-provider/sfw-provider-requests/src/responses.rs
+2 -2
View File
@@ -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<u8> {
sfw-provider/src/provider/client_handling/mod.rs
+3 -2
View File
@@ -154,8 +154,9 @@ impl ClientRequestProcessor {
}
fn generate_new_auth_token(data: Vec<u8>, 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());
sfw-provider/src/provider/mix_handling/mod.rs
+8 -2
View File
@@ -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<MixProcessingData>,
) -> Result<StoreData, MixProcessingError> {
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)) => {
sfw-provider/src/provider/mod.rs
+8 -1
View File
@@ -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);
sfw-provider/src/provider/storage/mod.rs
+27 -11
View File
@@ -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