From d71f860751d782da6d492b6ad6be5d4b7ff201bb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Tue, 3 May 2022 15:30:33 +0100 Subject: [PATCH] Changes to serialization + playing around with the concept of event dispatcher --- Cargo.lock | 10 +- .../coconut-dkg-contract/src/types.rs | 6 +- common/crypto/Cargo.toml | 9 ++ common/crypto/src/asymmetric/identity/mod.rs | 71 ++++++++++ common/crypto/src/lib.rs | 4 +- validator-api/Cargo.toml | 4 +- validator-api/src/dkg/dealing_processing.rs | 41 ++++++ validator-api/src/dkg/error.rs | 10 +- validator-api/src/dkg/events/dispatcher.rs | 39 ++++++ validator-api/src/dkg/events/event.rs | 10 ++ validator-api/src/dkg/events/mod.rs | 8 ++ validator-api/src/dkg/mod.rs | 2 + validator-api/src/dkg/networking/codec.rs | 8 +- validator-api/src/dkg/networking/handler.rs | 92 ++++++++++-- validator-api/src/dkg/networking/listener.rs | 2 +- validator-api/src/dkg/networking/message.rs | 131 +++++++++++------- validator-api/src/dkg/state.rs | 9 +- 17 files changed, 363 insertions(+), 93 deletions(-) create mode 100644 validator-api/src/dkg/dealing_processing.rs create mode 100644 validator-api/src/dkg/events/dispatcher.rs create mode 100644 validator-api/src/dkg/events/event.rs create mode 100644 validator-api/src/dkg/events/mod.rs diff --git a/Cargo.lock b/Cargo.lock index 4f04a75dd4..18cca302a4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1070,6 +1070,8 @@ dependencies = [ "pemstore", "rand 0.7.3", "rand_chacha 0.2.2", + "serde", + "serde_bytes", "subtle-encoding", "x25519-dalek", ] @@ -1434,6 +1436,7 @@ version = "1.4.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3d5c4b5e5959dc2c2b89918d8e2cc40fcdd623cef026ed09d2f0ee05199dc8e4" dependencies = [ + "serde", "signature", ] @@ -1447,6 +1450,7 @@ dependencies = [ "ed25519", "rand 0.7.3", "serde", + "serde_bytes", "sha2", "zeroize", ] @@ -3232,6 +3236,7 @@ dependencies = [ "anyhow", "async-trait", "attohttpc", + "bincode", "bytes", "cfg-if 1.0.0", "clap 2.34.0", @@ -3263,6 +3268,7 @@ dependencies = [ "rocket_cors", "rocket_sync_db_pools", "serde", + "serde_bytes", "serde_json", "sqlx", "thiserror", @@ -4836,9 +4842,9 @@ dependencies = [ [[package]] name = "serde_bytes" -version = "0.11.5" +version = "0.11.6" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "16ae07dd2f88a366f15bd0632ba725227018c69a1c8550a927324f8eb8368bb9" +checksum = "212e73464ebcde48d723aa02eb270ba62eff38a9b732df31f33f1b4e145f3a54" dependencies = [ "serde", ] diff --git a/common/cosmwasm-smart-contracts/coconut-dkg-contract/src/types.rs b/common/cosmwasm-smart-contracts/coconut-dkg-contract/src/types.rs index 8d208cdccf..caf9ad6507 100644 --- a/common/cosmwasm-smart-contracts/coconut-dkg-contract/src/types.rs +++ b/common/cosmwasm-smart-contracts/coconut-dkg-contract/src/types.rs @@ -9,7 +9,7 @@ pub type EncodedEd25519PublicKey = String; pub type EncodedEd25519PublicKeyRef<'a> = &'a str; pub type EncodedBTEPublicKeyWithProof = String; pub type NodeIndex = u64; -pub type EpochId = u64; +pub type EpochId = u32; #[derive(Serialize, Deserialize, Clone, Debug, PartialEq)] #[serde(rename_all = "snake_case")] @@ -89,7 +89,7 @@ impl Display for BlacklistingReason { } } -#[derive(Serialize, Deserialize, Clone, Debug, PartialEq)] +#[derive(Serialize, Deserialize, Clone, Copy, Debug, PartialEq)] #[serde(rename_all = "snake_case")] pub struct Epoch { pub id: EpochId, @@ -157,7 +157,7 @@ impl Epoch { // 6. VerificationKeyMismatchSubmission -> receivers / watchers raising issue that the submitted VK are mismatched with their local derivations // 7. VerificationKeyMismatchVoting -> (if any complaints were submitted) receivers voting on received mismatches // 8. InProgress -> all receivers have all their secrets derived and all is good -#[derive(Serialize, Deserialize, Clone, Debug, PartialEq)] +#[derive(Serialize, Deserialize, Clone, Copy, Debug, PartialEq)] #[serde(rename_all = "snake_case")] pub enum EpochState { PublicKeySubmission { diff --git a/common/crypto/Cargo.toml b/common/crypto/Cargo.toml index af5d24b904..5ae5901d25 100644 --- a/common/crypto/Cargo.toml +++ b/common/crypto/Cargo.toml @@ -26,10 +26,19 @@ nymsphinx-types = { path = "../nymsphinx/types" } pemstore = { path = "../../common/pemstore" } config = { path="../../common/config" } +serde_bytes = { version = "0.11.6", optional = true} + +[dependencies.serde_crate] +version = "1.0" +optional = true +default-features = false +package = "serde" + [dev-dependencies] rand_chacha = "0.2" [features] +serde = ["serde_crate", "serde_bytes", "ed25519-dalek/serde"] asymmetric = ["x25519-dalek", "ed25519-dalek"] hashing = ["blake3", "digest", "hkdf", "hmac", "generic-array"] symmetric = ["aes", "ctr", "cipher", "generic-array"] diff --git a/common/crypto/src/asymmetric/identity/mod.rs b/common/crypto/src/asymmetric/identity/mod.rs index d020d1953f..e3c21469e0 100644 --- a/common/crypto/src/asymmetric/identity/mod.rs +++ b/common/crypto/src/asymmetric/identity/mod.rs @@ -2,6 +2,7 @@ // SPDX-License-Identifier: Apache-2.0 pub use ed25519_dalek::ed25519::signature::Signature as SignatureTrait; +use ed25519_dalek::SecretKey; pub use ed25519_dalek::SignatureError; pub use ed25519_dalek::{Verifier, PUBLIC_KEY_LENGTH, SECRET_KEY_LENGTH, SIGNATURE_LENGTH}; use nymsphinx_types::{DestinationAddressBytes, DESTINATION_ADDRESS_LENGTH}; @@ -10,6 +11,13 @@ use pemstore::traits::{PemStorableKey, PemStorableKeyPair}; use rand::{CryptoRng, RngCore}; use std::fmt::{self, Display, Formatter}; +#[cfg(feature = "serde")] +use serde::de::Error as SerdeError; +#[cfg(feature = "serde")] +use serde::{Deserialize, Deserializer, Serialize, Serializer}; +#[cfg(feature = "serde")] +use serde_bytes::{ByteBuf as SerdeByteBuf, Bytes as SerdeBytes}; + #[derive(Debug)] pub enum Ed25519RecoveryError { MalformedBytes(SignatureError), @@ -135,6 +143,28 @@ impl PublicKey { } } +#[cfg(feature = "serde")] +impl Serialize for PublicKey { + fn serialize(&self, serializer: S) -> Result + where + S: Serializer, + { + self.0.serialize(serializer) + } +} + +#[cfg(feature = "serde")] +impl<'d> Deserialize<'d> for PublicKey { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'d>, + { + Ok(PublicKey(ed25519_dalek::PublicKey::deserialize( + deserializer, + )?)) + } +} + impl PemStorableKey for PublicKey { type Error = Ed25519RecoveryError; @@ -200,6 +230,26 @@ impl PrivateKey { } } +#[cfg(feature = "serde")] +impl Serialize for PrivateKey { + fn serialize(&self, serializer: S) -> Result + where + S: Serializer, + { + self.0.serialize(serializer) + } +} + +#[cfg(feature = "serde")] +impl<'d> Deserialize<'d> for PrivateKey { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'d>, + { + Ok(PrivateKey(SecretKey::deserialize(deserializer)?)) + } +} + impl PemStorableKey for PrivateKey { type Error = Ed25519RecoveryError; @@ -237,3 +287,24 @@ impl Signature { Ok(Signature(ed25519_dalek::Signature::from_bytes(bytes)?)) } } + +#[cfg(feature = "serde")] +impl Serialize for Signature { + fn serialize(&self, serializer: S) -> Result + where + S: Serializer, + { + SerdeBytes::new(&self.to_bytes()).serialize(serializer) + } +} + +#[cfg(feature = "serde")] +impl<'d> Deserialize<'d> for Signature { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'d>, + { + let bytes = ::deserialize(deserializer)?; + Signature::from_bytes(bytes.as_ref()).map_err(SerdeError::custom) + } +} diff --git a/common/crypto/src/lib.rs b/common/crypto/src/lib.rs index 456c4358fd..19aab595d4 100644 --- a/common/crypto/src/lib.rs +++ b/common/crypto/src/lib.rs @@ -29,5 +29,5 @@ pub use blake3; #[cfg(feature = "symmetric")] pub use ctr; -// TODO: this function uses all three modules: asymmetric crypto, symmetric crypto and derives key..., -// so I don't know where to put it... +#[cfg(feature = "serde")] +extern crate serde_crate as serde; diff --git a/validator-api/Cargo.toml b/validator-api/Cargo.toml index 4049770a98..e005d03607 100644 --- a/validator-api/Cargo.toml +++ b/validator-api/Cargo.toml @@ -44,15 +44,17 @@ rocket_sync_db_pools = { version = "0.1.0-rc.1", default-features = false } sqlx = { version = "0.5", features = ["runtime-tokio-rustls", "sqlite", "macros", "migrate"]} # required for DKG, perhaps will get locked behind feature flag +bincode = "1.3.3" bytes = "1.1.0" coconut-dkg-common = { path = "../common/cosmwasm-smart-contracts/coconut-dkg-contract" } dkg = { path = "../common/crypto/dkg" } +serde_bytes = { version = "0.11.6" } tokio-util = { version = "0.6", features = ["codec"] } ## internal coconut-bandwidth-contract-common = { path = "../common/cosmwasm-smart-contracts/coconut-bandwidth-contract" } config = { path = "../common/config" } -crypto = { path="../common/crypto" } +crypto = { path="../common/crypto", features = ["asymmetric", "serde"] } gateway-client = { path="../common/client-libs/gateway-client" } mixnet-contract-common = { path= "../common/cosmwasm-smart-contracts/mixnet-contract" } nymsphinx = { path="../common/nymsphinx" } diff --git a/validator-api/src/dkg/dealing_processing.rs b/validator-api/src/dkg/dealing_processing.rs new file mode 100644 index 0000000000..722c18186d --- /dev/null +++ b/validator-api/src/dkg/dealing_processing.rs @@ -0,0 +1,41 @@ +// Copyright 2022 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::dkg::networking::message::NewDealingMessage; +use crate::dkg::state::DkgState; +use futures::channel::mpsc; +use futures::StreamExt; +use log::error; + +// Once the DKG epoch begins, all parties will begin exchanging dealings with each other. +// We really don't want to be processing all of them in parallel since we would starve other +// parts of the process of CPU. Hence we put them in a queue and process each of them one by one. +pub(crate) type DealingReceiver = mpsc::UnboundedReceiver; +pub(crate) type DealingSender = mpsc::UnboundedSender; + +pub(crate) struct Processor { + dkg_state: DkgState, + receiver: DealingReceiver, +} +impl Processor { + pub(crate) fn new(dkg_state: DkgState, receiver: DealingReceiver) -> Self { + Processor { + dkg_state, + receiver, + } + } + + async fn process_dealing(&self, dealing: NewDealingMessage) { + todo!() + } + + pub(crate) async fn run(&mut self) { + while let Some(dealing) = self.receiver.next().await { + self.process_dealing(dealing).await + } + + // since we have no graceful shutdowns, seeing this error means something bad has happened + // as all senders got dropped + error!("") + } +} diff --git a/validator-api/src/dkg/error.rs b/validator-api/src/dkg/error.rs index 5e389919c5..ac9c586755 100644 --- a/validator-api/src/dkg/error.rs +++ b/validator-api/src/dkg/error.rs @@ -1,7 +1,6 @@ // Copyright 2022 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -use crate::dkg::networking::message::InvalidDkgMessageType; use std::io; use thiserror::Error; use validator_client::ValidatorClientError; @@ -17,12 +16,9 @@ pub enum DkgError { #[error("Networking error - {0}")] Networking(#[from] io::Error), + #[error("Failed to serialize message - {0}")] + SerializationError(#[from] bincode::Error), + #[error("todo")] DeserializationError, } - -impl From for DkgError { - fn from(err: InvalidDkgMessageType) -> Self { - todo!("figure out how it would fit in the DeserializationError") - } -} diff --git a/validator-api/src/dkg/events/dispatcher.rs b/validator-api/src/dkg/events/dispatcher.rs new file mode 100644 index 0000000000..d1616bf192 --- /dev/null +++ b/validator-api/src/dkg/events/dispatcher.rs @@ -0,0 +1,39 @@ +// Copyright 2022 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::dkg::dealing_processing::DealingSender; +use crate::dkg::events::Event; +use futures::channel::mpsc; +use futures::StreamExt; +use log::error; + +pub(crate) type DispatcherSender = mpsc::UnboundedSender; +pub(crate) type DispatcherReceiver = mpsc::UnboundedReceiver; + +pub(crate) struct Dispatcher { + event_receiver: DispatcherReceiver, + + dealing_processor: DealingSender, +} + +impl Dispatcher { + fn handle_event(&self, event: Event) { + match event { + Event::NewDealing(new_dealing_request) => self + .dealing_processor + .unbounded_send(new_dealing_request) + .expect("failed to forward new dealing message"), + _ => todo!(), + } + } + + pub(crate) async fn run(&mut self) { + while let Some(new_event) = self.event_receiver.next().await { + self.handle_event(new_event) + } + + // since we have no graceful shutdowns, seeing this error means something bad has happened + // as all senders got dropped + error!("") + } +} diff --git a/validator-api/src/dkg/events/event.rs b/validator-api/src/dkg/events/event.rs new file mode 100644 index 0000000000..dac1d66d5e --- /dev/null +++ b/validator-api/src/dkg/events/event.rs @@ -0,0 +1,10 @@ +// Copyright 2022 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::dkg::networking::message::{NewDealingMessage, RemoteDealingRequestMessage}; + +#[derive(Debug)] +pub(crate) enum Event { + NewDealing(NewDealingMessage), + NewDealingRequest(RemoteDealingRequestMessage), +} diff --git a/validator-api/src/dkg/events/mod.rs b/validator-api/src/dkg/events/mod.rs new file mode 100644 index 0000000000..32ffbaa5d5 --- /dev/null +++ b/validator-api/src/dkg/events/mod.rs @@ -0,0 +1,8 @@ +// Copyright 2022 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +mod dispatcher; +mod event; + +pub(crate) use dispatcher::DispatcherSender; +pub(crate) use event::Event; diff --git a/validator-api/src/dkg/mod.rs b/validator-api/src/dkg/mod.rs index f8c0cdc50f..dc3f32d490 100644 --- a/validator-api/src/dkg/mod.rs +++ b/validator-api/src/dkg/mod.rs @@ -5,3 +5,5 @@ mod contract_watcher; pub(crate) mod error; pub(crate) mod networking; pub(crate) mod state; +mod dealing_processing; +pub(crate) mod events; diff --git a/validator-api/src/dkg/networking/codec.rs b/validator-api/src/dkg/networking/codec.rs index 01513baecb..62ff1ea550 100644 --- a/validator-api/src/dkg/networking/codec.rs +++ b/validator-api/src/dkg/networking/codec.rs @@ -16,8 +16,7 @@ impl Encoder for DkgCodec { type Error = DkgError; fn encode(&mut self, item: OffchainDkgMessage, dst: &mut BytesMut) -> Result<(), Self::Error> { - item.encode(dst); - Ok(()) + item.encode(dst) } } @@ -54,9 +53,6 @@ impl Decoder for DkgCodec { 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(OffchainDkgMessage::try_from_bytes( - payload, - header.message_type, - )?)) + Ok(Some(OffchainDkgMessage::try_from_bytes(payload)?)) } } diff --git a/validator-api/src/dkg/networking/handler.rs b/validator-api/src/dkg/networking/handler.rs index c678f5ecb3..0f2369c22c 100644 --- a/validator-api/src/dkg/networking/handler.rs +++ b/validator-api/src/dkg/networking/handler.rs @@ -1,45 +1,118 @@ // Copyright 2022 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 +use crate::dkg::events::DispatcherSender; use crate::dkg::networking::codec::DkgCodec; -use crate::dkg::networking::message::{ErrorReason, OffchainDkgMessage}; +use crate::dkg::networking::message::{ + ErrorReason, NewDealingMessage, OffchainDkgMessage, RemoteDealingRequestMessage, +}; use crate::dkg::state::DkgState; -use futures::StreamExt; +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::new(2 * 60, 0); +const DEFAULT_MAX_CONNECTION_DURATION: Duration = Duration::new(2 * 60 * 60, 0); #[derive(Debug)] pub(crate) struct ConnectionHandler { // connection cannot exist for more than this time max_connection_duration: Duration, + dispatcher_sender: DispatcherSender, dkg_state: DkgState, conn: Framed, remote: SocketAddr, } impl ConnectionHandler { - pub(crate) fn new(dkg_state: DkgState, conn: TcpStream, remote: SocketAddr) -> Self { + pub(crate) fn new( + dispatcher_sender: DispatcherSender, + dkg_state: DkgState, + conn: TcpStream, + remote: SocketAddr, + ) -> Self { ConnectionHandler { max_connection_duration: DEFAULT_MAX_CONNECTION_DURATION, + dispatcher_sender, dkg_state, remote, conn: Framed::new(conn, DkgCodec), } } + async fn send_response(&mut self, response_message: OffchainDkgMessage) { + self.conn.send(response_message).await; + } + async fn send_error_response>( - &mut self, + &self, error: ErrorReason, additional_info: Option, ) { // } + async fn handle_new_dealing(&self, id: u64, message: NewDealingMessage) { + todo!() + } + + async fn handle_remote_dealing_request( + &mut self, + id: u64, + message: RemoteDealingRequestMessage, + ) { + // TODO: when somebody is reviewing this code, what's your opinion on accessing the DkgState here + // vs keeping it slightly more consistent and dispatching an event to request the value from + // something managing it instead? + + // personal note: once more parts are developed, I might change it myself before it even gets to the PR state + + let current_epoch = self.dkg_state.current_epoch().await; + if current_epoch.id != message.epoch_id { + return self + .send_error_response( + ErrorReason::InvalidEpoch, + Some(format!( + "current epoch is {} and not {}", + current_epoch.id, message.epoch_id + )), + ) + .await; + } + + let dealing = self.dkg_state.get_verified_dealing(message.dealer).await; + + self.send_response(todo!()).await; + todo!() + } + + async fn handle_request(&mut self, request: OffchainDkgMessage) { + match request { + OffchainDkgMessage::NewDealing { id, message } => { + self.handle_new_dealing(id, message).await + } + OffchainDkgMessage::RemoteDealingRequest { id, message } => { + self.handle_remote_dealing_request(id, message).await + } + OffchainDkgMessage::RemoteDealingResponse { .. } => { + self.send_error_response( + ErrorReason::InvalidRequest, + Some("RemoteDealingResponse is not a valid request type"), + ) + .await + } + OffchainDkgMessage::ErrorResponse { .. } => { + self.send_error_response( + ErrorReason::InvalidRequest, + Some("ErrorResponse is not a valid request type"), + ) + .await + } + } + } + async fn _handle_connection(&mut self) { debug!("Starting connection handler for {}", self.remote); @@ -92,13 +165,4 @@ impl ConnectionHandler { .await; } } - - async fn handle_request(&self, request: OffchainDkgMessage) { - match request { - OffchainDkgMessage::NewDealing { .. } => {} - OffchainDkgMessage::RemoteDealingRequest { .. } => {} - OffchainDkgMessage::RemoteDealingResponse { .. } => {} - OffchainDkgMessage::ErrorResponse { .. } => {} - } - } } diff --git a/validator-api/src/dkg/networking/listener.rs b/validator-api/src/dkg/networking/listener.rs index 081a31ec8d..80ed35d8f6 100644 --- a/validator-api/src/dkg/networking/listener.rs +++ b/validator-api/src/dkg/networking/listener.rs @@ -24,7 +24,7 @@ impl Listener { fn on_connect(&self, conn: TcpStream, remote: SocketAddr) { tokio::spawn( - ConnectionHandler::new(self.dkg_state.clone(), conn, remote).handle_connection(), + async move { todo!() }, // ConnectionHandler::new(self.dkg_state.clone(), conn, remote).handle_connection(), ); } diff --git a/validator-api/src/dkg/networking/message.rs b/validator-api/src/dkg/networking/message.rs index 46512758f3..9827863daa 100644 --- a/validator-api/src/dkg/networking/message.rs +++ b/validator-api/src/dkg/networking/message.rs @@ -2,11 +2,15 @@ // SPDX-License-Identifier: Apache-2.0 use crate::dkg::error::DkgError; +use crate::dkg::networking::PROTOCOL_VERSION; use bytes::{BufMut, BytesMut}; use crypto::asymmetric::identity; use dkg::Dealing; +use serde::{Deserialize, Serialize}; +use std::fmt::{Display, Formatter}; use std::io; +#[derive(Debug, Serialize, Deserialize)] pub enum OffchainDkgMessage { NewDealing { id: u64, @@ -26,82 +30,109 @@ pub enum OffchainDkgMessage { }, } +#[derive(Debug, Serialize, Deserialize)] pub struct NewDealingMessage { - epoch_id: u32, + pub epoch_id: u32, // we keep the dealing in its serialized state as that's what is being signed (and hashed) // so that it's easier to verify - dealing_bytes: Vec, - dealer_signature: identity::Signature, + pub dealing_bytes: Vec, + pub dealer_signature: identity::Signature, } +#[derive(Debug, Serialize, Deserialize)] pub struct RemoteDealingRequestMessage { - epoch_id: u32, - dealer: identity::PublicKey, + pub epoch_id: u32, + pub dealer: identity::PublicKey, } +#[derive(Debug, Serialize, Deserialize)] pub enum RemoteDealingResponseMessage { Available { epoch_id: u32, + #[serde(with = "dealing_bytes")] dealing: Box, dealer_signature: identity::Signature, }, Unavailable, } -pub struct ErrorResponseMessage { - reason: ErrorReason, - additional_info: Option, +mod dealing_bytes { + use dkg::Dealing; + use serde::de::Error as SerdeError; + use serde::{Deserialize, Deserializer, Serialize, Serializer}; + use serde_bytes::{ByteBuf as SerdeByteBuf, Bytes as SerdeBytes}; + + pub fn serialize(val: &Dealing, serializer: S) -> Result { + SerdeBytes::new(&val.to_bytes()).serialize(serializer) + } + + pub fn deserialize<'de, D: Deserializer<'de>>( + deserializer: D, + ) -> Result, D::Error> { + let bytes = ::deserialize(deserializer)?; + let dealing = Dealing::try_from_bytes(bytes.as_ref()).map_err(SerdeError::custom)?; + Ok(Box::new(dealing)) + } } +#[derive(Debug, Serialize, Deserialize)] +pub struct ErrorResponseMessage { + pub reason: ErrorReason, + pub additional_info: Option, +} + +impl ErrorResponseMessage { + pub fn new(reason: ErrorReason, additional_info: Option) -> Self { + ErrorResponseMessage { + reason, + additional_info, + } + } +} + +impl Display for ErrorResponseMessage { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + todo!() + } +} + +#[derive(Debug, Serialize, Deserialize)] pub enum ErrorReason { + InvalidEpoch, UnknownDealer, + InvalidRequest, Timeout, } +impl Display for ErrorReason { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + todo!() + } +} + impl OffchainDkgMessage { - fn frame(self) -> FramedOffchainDkgMessage { - todo!() + pub(crate) fn try_from_bytes(bytes: Vec) -> Result { + Ok(bincode::deserialize(&bytes)?) } - pub(crate) fn encode(self, dst: &mut BytesMut) { - dst.put(self.frame().into_bytes().as_ref()); + pub(crate) fn try_to_bytes(&self) -> Result, DkgError> { + Ok(bincode::serialize(&self)?) } - pub(crate) fn try_from_bytes( - bytes: Vec, - expected_type: OffchainDkgMessageType, - ) -> Result { - todo!() + 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, + }) } -} -#[repr(u8)] -#[derive(Clone, Copy, Debug, PartialEq)] -pub(crate) enum OffchainDkgMessageType { - NewDealing = 0, - RemoteDealingRequest = 1, - - RemoteDealingResponse = 128, - ErrorResponse = 255, -} - -pub struct InvalidDkgMessageType(u8); - -impl TryFrom for OffchainDkgMessageType { - type Error = InvalidDkgMessageType; - - fn try_from(value: u8) -> std::result::Result { - match value { - _ if value == (OffchainDkgMessageType::NewDealing as u8) => Ok(Self::NewDealing), - _ if value == (OffchainDkgMessageType::RemoteDealingRequest as u8) => { - Ok(Self::RemoteDealingRequest) - } - _ if value == (OffchainDkgMessageType::RemoteDealingResponse as u8) => { - Ok(Self::RemoteDealingResponse) - } - _ if value == (OffchainDkgMessageType::ErrorResponse as u8) => Ok(Self::ErrorResponse), - t => Err(InvalidDkgMessageType(t)), - } + pub(crate) fn encode(self, dst: &mut BytesMut) -> Result<(), DkgError> { + dst.put(self.frame()?.into_bytes().as_ref()); + Ok(()) } } @@ -127,17 +158,15 @@ impl FramedOffchainDkgMessage { #[derive(Debug, Copy, Clone, PartialEq)] pub(crate) struct Header { - pub(crate) message_type: OffchainDkgMessageType, pub(crate) payload_length: u64, pub(crate) protocol_version: u32, } impl Header { - pub(crate) const LEN: usize = 13; + pub(crate) const LEN: usize = 12; pub(crate) fn into_bytes(self) -> Vec { let mut out = Vec::with_capacity(Self::LEN); - out.push(self.message_type as u8); out.extend_from_slice(&self.payload_length.to_be_bytes()); out.extend_from_slice(&self.protocol_version.to_be_bytes()); @@ -157,9 +186,8 @@ impl Header { ))); } Ok(Header { - message_type: OffchainDkgMessageType::try_from(bytes[0])?, - payload_length: u64::from_be_bytes(bytes[1..9].try_into().unwrap()), - protocol_version: u32::from_be_bytes(bytes[9..].try_into().unwrap()), + payload_length: u64::from_be_bytes(bytes[..8].try_into().unwrap()), + protocol_version: u32::from_be_bytes(bytes[8..].try_into().unwrap()), }) } } @@ -172,7 +200,6 @@ mod tests { #[test] fn header_deserialization() { let valid_header = Header { - message_type: OffchainDkgMessageType::NewDealing, payload_length: 1234, protocol_version: PROTOCOL_VERSION, }; diff --git a/validator-api/src/dkg/state.rs b/validator-api/src/dkg/state.rs index cf7c441744..504ccf330c 100644 --- a/validator-api/src/dkg/state.rs +++ b/validator-api/src/dkg/state.rs @@ -42,7 +42,6 @@ struct DkgStateInner { impl DkgState { // some save/load action here - pub(crate) async fn is_dealers_remote_address(&self, remote: SocketAddr) -> bool { let dealers = &self.inner.lock().await.current_epoch_dealers; dealers @@ -50,6 +49,10 @@ impl DkgState { .any(|dealer| dealer.remote_address == remote) } + pub(crate) async fn current_epoch(&self) -> Epoch { + self.inner.lock().await.current_epoch + } + pub(crate) async fn get_verified_dealing( &self, dealer: identity::PublicKey, @@ -62,7 +65,3 @@ impl DkgState { .cloned() } } - -impl DkgStateInner { - // -}