diff --git a/validator-api/src/main.rs b/validator-api/src/main.rs index 461d11ab51..99313f13e0 100644 --- a/validator-api/src/main.rs +++ b/validator-api/src/main.rs @@ -45,6 +45,9 @@ mod rewarded_set_updater; pub(crate) mod storage; mod swagger; +#[allow(dead_code)] +mod networking; + #[cfg(feature = "coconut")] mod coconut; diff --git a/validator-api/src/networking/codec.rs b/validator-api/src/networking/codec.rs new file mode 100644 index 0000000000..438e84040c --- /dev/null +++ b/validator-api/src/networking/codec.rs @@ -0,0 +1,65 @@ +// Copyright 2022 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::networking::error::NetworkingError; +use crate::networking::message::{Header, OffchainMessage}; +use crate::networking::PROTOCOL_VERSION; +use bytes::{Buf, BytesMut}; +use tokio_util::codec::{Decoder, Encoder}; + +// TODO: that was fine for the purposes of DKG, we might want to increase it if it's used anywhere else +const MAX_ALLOWED_MESSAGE_LEN: u64 = 2 * 1024 * 1024; + +#[derive(Debug)] +pub struct OffchainCodec; + +impl<'a> Encoder<&'a OffchainMessage> for OffchainCodec { + type Error = NetworkingError; + + fn encode(&mut self, item: &OffchainMessage, dst: &mut BytesMut) -> Result<(), Self::Error> { + item.encode(dst) + } +} + +impl Decoder for OffchainCodec { + type Item = OffchainMessage; + type Error = NetworkingError; + + fn decode(&mut self, src: &mut BytesMut) -> Result, Self::Error> { + if src.len() < Header::LEN { + // can't do much without being able to deserialize the header + return Ok(None); + } + + // header deserialization is simple enough to not be too cumbersome if it were to be repeated + let header = Header::try_from_bytes(&src[..Header::LEN])?; + + if header.payload_length > MAX_ALLOWED_MESSAGE_LEN { + return Err(NetworkingError::MessageTooLarge { + supported: MAX_ALLOWED_MESSAGE_LEN, + received: header.payload_length, + }); + } + + if header.protocol_version != PROTOCOL_VERSION { + return Err(NetworkingError::MismatchedProtocolVersion { + expected: PROTOCOL_VERSION, + received: header.protocol_version, + }); + } + + if src.len() < Header::LEN + header.payload_length as usize { + // we haven't received the entire expected message yet. + // However, reserve enough bytes in the buffer for it + src.reserve(Header::LEN + header.payload_length as usize - src.len()); + + // We inform the Framed that we need more bytes to form the next frame. + return Ok(None); + } + + let payload = src[Header::LEN..Header::LEN + header.payload_length as usize].to_vec(); + src.advance(Header::LEN + header.payload_length as usize); + + Ok(Some(OffchainMessage::try_from_bytes(payload)?)) + } +} diff --git a/validator-api/src/networking/error.rs b/validator-api/src/networking/error.rs new file mode 100644 index 0000000000..9b53f58eb4 --- /dev/null +++ b/validator-api/src/networking/error.rs @@ -0,0 +1,22 @@ +// Copyright 2022 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use std::io; +use thiserror::Error; + +// this would probably need to adjusted or maybe incorporated into existing stuff as originally +// this error was more extensive by being a global `DkgError` +#[derive(Error, Debug)] +pub enum NetworkingError { + #[error("Networking / IO error - {0}")] + Io(#[from] io::Error), + + #[error("Received message with specified size bigger than the supported maximum. Received: {received}, supported: {supported}")] + MessageTooLarge { supported: u64, received: u64 }, + + #[error("Received message with unexpected protocol version. Received: {received}, expected: {expected}")] + MismatchedProtocolVersion { expected: u32, received: u32 }, + + #[error("Failed to deal with serialization (or deserialization) of the message - {0}")] + SerializationError(#[from] bincode::Error), +} diff --git a/validator-api/src/networking/message.rs b/validator-api/src/networking/message.rs new file mode 100644 index 0000000000..94c6219fd0 --- /dev/null +++ b/validator-api/src/networking/message.rs @@ -0,0 +1,165 @@ +// Copyright 2022 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::networking::error::NetworkingError; +use crate::networking::PROTOCOL_VERSION; +use bytes::{BufMut, BytesMut}; +use serde::{Deserialize, Serialize}; +use std::fmt::{Display, Formatter}; +use std::io; +use std::time::Duration; +use thiserror::Error; + +// I left a sample `NewDealingMessage` to show how it was originally implemented + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub enum OffchainMessage { + // you'd add something like this: + // NewDealing { + // id: u64, + // message: NewDealingMessage, + // }, + ErrorResponse { + id: Option, + message: ErrorResponseMessage, + }, +} + +impl Display for OffchainMessage { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + write!(f, "OffchainMessage ")?; + match self { + // OffchainDkgMessage::NewDealing { id, message } => { + // write!(f, "with id {} and message: {}", id, message) + // } + OffchainMessage::ErrorResponse { id, message } => { + write!(f, "with id {:?} and message: {}", id, message) + } + } + } +} + +// #[derive(Debug, Clone, Serialize, Deserialize)] +// pub struct NewDealingMessage { +// pub epoch_id: u32, +// pub dealing_bytes: Vec, +// pub dealer_signature: identity::Signature, +// } +// impl Display for NewDealingMessage { +// fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { +// write!( +// f, +// "NewDealingMessage for epoch {} with length {}", +// self.epoch_id, +// self.dealing_bytes.len() +// ) +// } +// } + +#[derive(Debug, Clone, Serialize, Deserialize, Error)] +pub enum ErrorResponseMessage { + #[error("{typ} is not a valid request type")] + InvalidRequest { typ: String }, + + #[error("This request failed to get resolved within {} seconds", .timeout.as_secs())] + Timeout { timeout: Duration }, +} + +impl OffchainMessage { + pub(crate) fn new_error_response(id: Option, message: ErrorResponseMessage) -> Self { + OffchainMessage::ErrorResponse { id, message } + } + + pub(crate) fn try_from_bytes(bytes: Vec) -> Result { + Ok(bincode::deserialize(&bytes)?) + } + + pub(crate) fn try_to_bytes(&self) -> Result, NetworkingError> { + Ok(bincode::serialize(&self)?) + } + + fn frame(&self) -> Result { + let payload = self.try_to_bytes()?; + Ok(FramedOffchainDkgMessage { + header: Header { + payload_length: payload.len() as u64, + protocol_version: PROTOCOL_VERSION, + }, + payload, + }) + } + + pub(crate) fn encode(&self, dst: &mut BytesMut) -> Result<(), NetworkingError> { + dst.put(self.frame()?.into_bytes().as_ref()); + Ok(()) + } +} + +struct FramedOffchainDkgMessage { + header: Header, + payload: Vec, +} + +impl FramedOffchainDkgMessage { + fn into_bytes(mut self) -> Vec { + let mut header_bytes = self.header.into_bytes(); + let mut out = Vec::with_capacity(header_bytes.len() + self.payload.len()); + + out.append(&mut header_bytes); + out.append(&mut self.payload); + out + } +} + +#[derive(Debug, Copy, Clone, PartialEq)] +pub(crate) struct Header { + pub(crate) payload_length: u64, + pub(crate) protocol_version: u32, +} + +impl Header { + pub(crate) const LEN: usize = 12; + + pub(crate) fn into_bytes(self) -> Vec { + let mut out = Vec::with_capacity(Self::LEN); + out.extend_from_slice(&self.payload_length.to_be_bytes()); + out.extend_from_slice(&self.protocol_version.to_be_bytes()); + + debug_assert_eq!(Self::LEN, out.len()); + out + } + + pub(crate) fn try_from_bytes(bytes: &[u8]) -> Result { + if bytes.len() != Self::LEN { + return Err(NetworkingError::Io(io::Error::new( + io::ErrorKind::InvalidData, + format!( + "OffchainMessageType::Header: got {} bytes, expected: {}", + bytes.len(), + Self::LEN + ), + ))); + } + Ok(Header { + payload_length: u64::from_be_bytes(bytes[..8].try_into().unwrap()), + protocol_version: u32::from_be_bytes(bytes[8..].try_into().unwrap()), + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::networking::PROTOCOL_VERSION; + + #[test] + fn header_deserialization() { + let valid_header = Header { + payload_length: 1234, + protocol_version: PROTOCOL_VERSION, + }; + + let bytes = valid_header.into_bytes(); + assert_eq!(valid_header, Header::try_from_bytes(&bytes).unwrap()) + } +} diff --git a/validator-api/src/networking/mod.rs b/validator-api/src/networking/mod.rs new file mode 100644 index 0000000000..3687c772bb --- /dev/null +++ b/validator-api/src/networking/mod.rs @@ -0,0 +1,14 @@ +// Copyright 2022 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +// TODO: if this becomes too cumbersome, perhaps consider a more streamlined solution like tarpc +// (I wouldn't have needed that for DKG, but if this is to be used for different purposes, maybe +// it would have been more appropriate) + +pub(crate) mod codec; +pub(crate) mod error; +pub(crate) mod message; +pub(crate) mod receiver; +pub(crate) mod sender; + +pub(crate) const PROTOCOL_VERSION: u32 = 1; diff --git a/validator-api/src/networking/receiver/handler.rs b/validator-api/src/networking/receiver/handler.rs new file mode 100644 index 0000000000..8bf0f914fe --- /dev/null +++ b/validator-api/src/networking/receiver/handler.rs @@ -0,0 +1,98 @@ +// Copyright 2022 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::networking::codec::OffchainCodec; +use crate::networking::message::{ErrorResponseMessage, OffchainMessage}; +// use crate::dkg::state::StateAccessor; +use futures::{SinkExt, StreamExt}; +use std::net::SocketAddr; +use std::time::Duration; +use tokio::net::TcpStream; +use tokio::time::timeout; +use tokio_util::codec::Framed; + +const DEFAULT_MAX_CONNECTION_DURATION: Duration = Duration::from_secs(2 * 60 * 60); + +#[derive(Debug)] +pub struct ConnectionHandler { + // connection cannot exist for more than this time + max_connection_duration: Duration, + // state_accessor: StateAccessor, + conn: Framed, + remote: SocketAddr, +} + +impl ConnectionHandler { + pub(crate) fn new(conn: TcpStream, remote: SocketAddr) -> Self { + ConnectionHandler { + max_connection_duration: DEFAULT_MAX_CONNECTION_DURATION, + remote, + conn: Framed::new(conn, OffchainCodec), + } + } + + async fn send_response(&mut self, response_message: OffchainMessage) { + if let Err(err) = self.conn.send(&response_message).await { + warn!("Failed to send response back to {} - {}", self.remote, err) + } + } + + async fn send_error_response(&mut self, id: Option, error: ErrorResponseMessage) { + self.send_response(OffchainMessage::new_error_response(id, error)) + .await + } + + async fn handle_request(&mut self, request: OffchainMessage) { + match request { + OffchainMessage::ErrorResponse { id, .. } => { + self.send_error_response( + id, + ErrorResponseMessage::InvalidRequest { + typ: "ErrorResponse".into(), + }, + ) + .await + } + } + } + + async fn _handle_connection(&mut self) { + debug!("Starting connection handler for {}", self.remote); + + while let Some(framed_dkg_request) = self.conn.next().await { + trace!("received new message from {}", self.remote); + match framed_dkg_request { + Ok(framed_dkg_request) => self.handle_request(framed_dkg_request).await, + Err(err) => { + warn!( + "The socket connection got corrupted with error: {:?}. Closing the socket", + err + ); + break; + } + } + } + + debug!("Closing connection from {}", self.remote); + } + + pub async fn handle_connection(mut self) { + let remote = self.remote; + if timeout(self.max_connection_duration, self._handle_connection()) + .await + .is_err() + { + warn!( + "we timed out while trying to resolve connection from {}", + remote + ); + self.send_error_response( + None, + ErrorResponseMessage::Timeout { + timeout: self.max_connection_duration, + }, + ) + .await; + } + } +} diff --git a/validator-api/src/networking/receiver/listener.rs b/validator-api/src/networking/receiver/listener.rs new file mode 100644 index 0000000000..9db7b1e2ef --- /dev/null +++ b/validator-api/src/networking/receiver/listener.rs @@ -0,0 +1,47 @@ +// Copyright 2022 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::networking::receiver::handler::ConnectionHandler; +use log::debug; +use std::fmt::Display; +use std::net::SocketAddr; +use std::process; +use tokio::net::{TcpListener, TcpStream, ToSocketAddrs}; + +// note that we do not expect persistent connections between dealers, they should only really +// exist for the duration of a single message exchange +pub struct Listener { + address: A, +} + +impl Listener { + pub(crate) fn new(address: A) -> Self { + Listener { address } + } + + fn on_connect(&self, conn: TcpStream, remote: SocketAddr) { + tokio::spawn(ConnectionHandler::new(conn, remote).handle_connection()); + } + + pub(crate) async fn run(&mut self) + where + A: ToSocketAddrs + Display, + { + debug!("starting off-chain DKG Listener"); + + let listener = match TcpListener::bind(&self.address).await { + Ok(listener) => listener, + Err(err) => { + error!("Failed to bind to {} - {}.", self.address, err); + process::exit(1); + } + }; + + loop { + match listener.accept().await { + Ok((socket, remote_addr)) => self.on_connect(socket, remote_addr), + Err(err) => warn!("Failed to accept incoming connection - {:?}", err), + } + } + } +} diff --git a/validator-api/src/networking/receiver/mod.rs b/validator-api/src/networking/receiver/mod.rs new file mode 100644 index 0000000000..73b14e31c1 --- /dev/null +++ b/validator-api/src/networking/receiver/mod.rs @@ -0,0 +1,7 @@ +// Copyright 2022 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +mod handler; +mod listener; + +pub(crate) use listener::Listener; diff --git a/validator-api/src/networking/sender/broadcast.rs b/validator-api/src/networking/sender/broadcast.rs new file mode 100644 index 0000000000..9c5c4a7692 --- /dev/null +++ b/validator-api/src/networking/sender/broadcast.rs @@ -0,0 +1,122 @@ +// Copyright 2022 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::networking::message::OffchainMessage; +use crate::networking::sender::{send_single_message, ConnectionConfig, SendResponse}; +use futures::channel::mpsc; +use futures::{stream, StreamExt}; +use std::net::SocketAddr; + +// TODO: for now just leave it here and make it configurable with proper config later +const DEFAULT_CONCURRENCY: usize = 5; + +type FeedbackSender = mpsc::UnboundedSender; + +pub(crate) struct Broadcaster { + addresses: Vec, + concurrency_level: usize, + config: ConnectionConfig, +} + +impl Broadcaster { + pub(crate) fn new(addresses: Vec, config: ConnectionConfig) -> Self { + Broadcaster { + addresses, + concurrency_level: DEFAULT_CONCURRENCY, + config, + } + } + + pub(crate) fn with_concurrency_level(mut self, concurrency_level: usize) -> Self { + self.concurrency_level = concurrency_level; + self + } + + pub(crate) fn set_addresses(&mut self, new_addresses: Vec) { + self.addresses = new_addresses; + } + + fn create_broadcast_configs( + &self, + message: OffchainMessage, + feedback_sender: Option, + ) -> Vec { + self.addresses + .iter() + .map(|&address| BroadcastConfig { + address, + config: self.config, + feedback_sender: feedback_sender.clone(), + message: message.clone(), + }) + .collect() + } + + pub(crate) async fn broadcast_with_feedback(&self, msg: OffchainMessage) -> Vec { + if self.addresses.is_empty() { + warn!("attempting to broadcast {} while no remotes are known", msg); + return Vec::new(); + } + + debug!("broadcasting {} to {} remotes", msg, self.addresses.len()); + let (feedback_tx, mut feedback_rx) = mpsc::unbounded(); + + stream::iter(self.create_broadcast_configs(msg, Some(feedback_tx))) + .for_each_concurrent(self.concurrency_level, |cfg| cfg.send()) + .await; + + let mut responses = Vec::new(); + + for _ in 0..self.addresses.len() { + // we should have received exactly self.addresses number of responses + // (they could be just Err failure responses, but should exist nonetheless) + match feedback_rx.try_next() { + Ok(Some(response)) => responses.push(response), + Err(_) | Ok(None) => { + error!("somehow we received fewer feedback responses than sent messages") + } + } + } + + // the channel should have been drained and all sender should have been dropped + debug_assert!(matches!(feedback_rx.try_next(), Ok(None))); + responses + } + + pub(crate) async fn broadcast(&self, msg: OffchainMessage) { + if self.addresses.is_empty() { + warn!("attempting to broadcast {} while no remotes are known", msg); + return; + } + + debug!("broadcasting {} to {} remotes", msg, self.addresses.len()); + stream::iter(self.create_broadcast_configs(msg, None)) + .for_each_concurrent(self.concurrency_level, |cfg| cfg.send()) + .await + } +} + +// internal struct to have per-connection config on hand +struct BroadcastConfig { + address: SocketAddr, + config: ConnectionConfig, + feedback_sender: Option, + message: OffchainMessage, +} + +impl BroadcastConfig { + async fn send(self) { + let response = send_single_message(self.address, self.config, &self.message).await; + if let Some(feedback_sender) = self.feedback_sender { + // this can only fail if the receiver is disconnected which should never be the case + // thus we can ignore the possible error + let _ = feedback_sender.unbounded_send(response); + } else if let Err(err) = response.response { + // if we're not forwarding feedback, at least emit a warning about the failure + warn!( + "failed to broadcast {} to {} - {}", + self.message, self.address, err + ) + } + } +} diff --git a/validator-api/src/networking/sender/ephemeral.rs b/validator-api/src/networking/sender/ephemeral.rs new file mode 100644 index 0000000000..25156bc13c --- /dev/null +++ b/validator-api/src/networking/sender/ephemeral.rs @@ -0,0 +1,83 @@ +// Copyright 2022 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::networking::codec::OffchainCodec; +use crate::networking::error::NetworkingError; +use crate::networking::message::OffchainMessage; +use crate::networking::sender::ConnectionConfig; +use futures::{SinkExt, StreamExt}; +use std::io; +use std::net::SocketAddr; +use std::time::Duration; +use tokio::net::TcpStream; +use tokio::time::timeout; +use tokio_util::codec::Framed; + +// this connection only exists for a single message +pub(crate) struct EphemeralConnection { + remote: SocketAddr, + conn: Framed, +} + +impl EphemeralConnection { + pub(crate) async fn connect( + address: SocketAddr, + connection_timeout: Duration, + ) -> io::Result { + trace!("attempting to connect to {}", address); + let conn = match timeout(connection_timeout, TcpStream::connect(address)).await { + Err(_timeout) => { + return Err(io::Error::new( + io::ErrorKind::Other, + format!("timed out while attempting to send message to {}", address), + )) + } + Ok(conn_res) => conn_res?, + }; + let framed_conn = Framed::new(conn, OffchainCodec); + Ok(Self { + remote: address, + conn: framed_conn, + }) + } + + pub(crate) fn remote(&self) -> SocketAddr { + self.remote + } + + async fn send( + &mut self, + message: &OffchainMessage, + send_timeout: Duration, + response_timeout: Option, + ) -> Result, NetworkingError> { + trace!("attempting to send to {}", self.remote); + match timeout(send_timeout, self.conn.send(message)).await { + Err(_timeout) => { + return Err(NetworkingError::Io(io::Error::new( + io::ErrorKind::Other, + "timed out while attempting to send message", + ))) + } + Ok(res) => res?, + } + if let Some(response_timeout) = response_timeout { + match timeout(response_timeout, self.conn.next()).await { + Err(_elapsed) => Ok(None), + Ok(response) => response.transpose(), + } + } else { + Ok(None) + } + } + + pub(crate) async fn connect_and_send( + address: SocketAddr, + cfg: ConnectionConfig, + message: &OffchainMessage, + ) -> Result, NetworkingError> { + let mut conn = EphemeralConnection::connect(address, cfg.connection_timeout).await?; + conn.send(message, cfg.send_timeout, cfg.response_timeout) + .await + } +} diff --git a/validator-api/src/networking/sender/mod.rs b/validator-api/src/networking/sender/mod.rs new file mode 100644 index 0000000000..0aa21689a7 --- /dev/null +++ b/validator-api/src/networking/sender/mod.rs @@ -0,0 +1,65 @@ +// Copyright 2022 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::networking::error::NetworkingError; +use crate::networking::message::OffchainMessage; +use crate::networking::sender::broadcast::Broadcaster; +use crate::networking::sender::ephemeral::EphemeralConnection; +use std::net::SocketAddr; +use std::time::Duration; + +pub(crate) mod broadcast; +pub(crate) mod ephemeral; + +#[derive(Debug, Clone, Copy)] +pub(crate) struct ConnectionConfig { + pub(crate) connection_timeout: Duration, + pub(crate) response_timeout: Option, + pub(crate) send_timeout: Duration, +} + +pub(crate) struct SendResponse { + pub(crate) source: SocketAddr, + pub(crate) response: Result, NetworkingError>, +} + +impl SendResponse { + pub(crate) fn new( + source: SocketAddr, + response: Result, NetworkingError>, + ) -> Self { + SendResponse { source, response } + } +} + +pub(crate) async fn send_single_message( + address: SocketAddr, + cfg: ConnectionConfig, + message: &OffchainMessage, +) -> SendResponse { + let res = EphemeralConnection::connect_and_send(address, cfg, message).await; + SendResponse::new(address, res) +} + +pub(crate) async fn broadcast_message( + addresses: Vec, + cfg: ConnectionConfig, + message: &OffchainMessage, +) { + Broadcaster::new(addresses, cfg) + .broadcast(message.clone()) + .await +} + +pub(crate) async fn broadcast_message_with_feedback( + addresses: Vec, + cfg: ConnectionConfig, + message: &OffchainMessage, +) -> Vec { + Broadcaster::new(addresses, cfg) + .broadcast_with_feedback(message.clone()) + .await +} + +// NOTE: for the original purposes of DKG stateless broadcasts and one-off sends were enough, +// so I never implemented proper persistent connections