diff --git a/Cargo.lock b/Cargo.lock index a3a8ad28f3..40198fbcaa 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -735,6 +735,8 @@ dependencies = [ "blake3", "cosmwasm-std", "digest 0.10.3", + "schemars", + "serde", ] [[package]] diff --git a/common/cosmwasm-smart-contracts/coconut-dkg-contract/src/msg.rs b/common/cosmwasm-smart-contracts/coconut-dkg-contract/src/msg.rs index ecdb36babe..22043a6676 100644 --- a/common/cosmwasm-smart-contracts/coconut-dkg-contract/src/msg.rs +++ b/common/cosmwasm-smart-contracts/coconut-dkg-contract/src/msg.rs @@ -1,7 +1,7 @@ // Copyright 2022 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -use crate::types::{BlockHeight, EncodedBTEPublicKeyWithProof, EncodedEd25519PublicKey}; +use crate::types::{BlockHeight, EncodedBTEPublicKeyWithProof, EncodedEd25519PublicKey, Threshold}; use contracts_common::commitment::ContractSafeCommitment; use schemars::JsonSchema; use serde::{Deserialize, Serialize}; @@ -10,6 +10,7 @@ use serde::{Deserialize, Serialize}; #[serde(rename_all = "snake_case")] pub struct InstantiateMsg { pub public_key_submission_end_height: BlockHeight, + pub system_threshold: Option, } #[derive(Serialize, Deserialize, Clone, Debug, PartialEq, JsonSchema)] 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 8320e2ad0b..00d2c49135 100644 --- a/common/cosmwasm-smart-contracts/coconut-dkg-contract/src/types.rs +++ b/common/cosmwasm-smart-contracts/coconut-dkg-contract/src/types.rs @@ -7,6 +7,7 @@ pub use crate::dealer::{ BlacklistedDealer, Blacklisting, BlacklistingReason, BlacklistingResponse, DealerDetails, PagedBlacklistingResponse, PagedDealerResponse, }; +pub use contracts_common::commitment::ContractSafeCommitment; pub use cosmwasm_std::{Addr, Coin}; pub type BlockHeight = u64; @@ -15,6 +16,7 @@ pub type EncodedEd25519PublicKeyRef<'a> = &'a str; pub type EncodedBTEPublicKeyWithProof = String; pub type EncodedBTEPublicKeyWithProofRef<'a> = &'a str; pub type NodeIndex = u64; +pub type Threshold = u64; pub type EpochId = u32; #[derive(Serialize, Deserialize, Clone, Copy, Debug, PartialEq)] @@ -22,6 +24,9 @@ pub type EpochId = u32; pub struct Epoch { pub id: EpochId, pub state: EpochState, + + // TODO: need to ponder a bit whether it's actually a property of a particular epoch + pub system_threshold: Threshold, } impl Epoch { @@ -71,7 +76,11 @@ impl Epoch { }, }; - Some(Epoch { id: self.id, state }) + Some(Epoch { + id: self.id, + state, + system_threshold: self.system_threshold, + }) } } diff --git a/common/cosmwasm-smart-contracts/contracts-common/Cargo.toml b/common/cosmwasm-smart-contracts/contracts-common/Cargo.toml index 7ce30155c9..4de97b8045 100644 --- a/common/cosmwasm-smart-contracts/contracts-common/Cargo.toml +++ b/common/cosmwasm-smart-contracts/contracts-common/Cargo.toml @@ -8,8 +8,11 @@ edition = "2021" [dependencies] cosmwasm-std = "1.0.0-beta8" +schemars = "0.8" +serde = { version = "1.0.103", default-features = false, features = ["derive"] } -blake3 = { version = "1.3.1", features = ["traits-preview"], optional = true} + +blake3 = { version = "1.3.1", features = ["traits-preview"], optional = true } digest = { version = "0.10.3", optional = true } [features] diff --git a/common/cosmwasm-smart-contracts/contracts-common/src/commitment.rs b/common/cosmwasm-smart-contracts/contracts-common/src/commitment.rs index 4ba5e9c909..df565ac9f6 100644 --- a/common/cosmwasm-smart-contracts/contracts-common/src/commitment.rs +++ b/common/cosmwasm-smart-contracts/contracts-common/src/commitment.rs @@ -3,8 +3,12 @@ #[cfg(feature = "committable_trait")] pub use digest::{Digest, Output}; +use schemars::JsonSchema; +use serde::{Deserialize, Serialize}; +use std::fmt::{Display, Formatter}; #[cfg(feature = "committable_trait")] use std::marker::PhantomData; +use std::ops::Deref; // some sane upper-bound size on commitment sizes // currently set to 1024bits @@ -12,7 +16,34 @@ pub const MAX_COMMITMENT_SIZE: usize = 128; // TODO: if we are to use commitments for different types, it might make sense to introduce something like // CommitmentTypeId field on the below for distinguishing different ones. it would somehow become part of the trait -pub type ContractSafeCommitment = Vec; +#[derive( + Debug, Clone, Ord, PartialOrd, Eq, PartialEq, Hash, Serialize, Deserialize, JsonSchema, +)] +pub struct ContractSafeCommitment(Vec); + +impl Deref for ContractSafeCommitment { + type Target = Vec; + + fn deref(&self) -> &Self::Target { + &self.0 + } +} + +impl Display for ContractSafeCommitment { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + if !self.0.is_empty() { + write!(f, "0x")?; + } + for byte in self.0.iter().take(MAX_COMMITMENT_SIZE) { + write!(f, "{:02X}", byte)?; + } + // just some sanity safeguards + if self.0.len() > MAX_COMMITMENT_SIZE { + write!(f, "...")?; + } + Ok(()) + } +} #[derive(Debug, Copy, Clone, Ord, PartialOrd, Eq, PartialEq, Hash)] pub struct UnsupportedCommitmentSize; @@ -87,7 +118,7 @@ where T: ?Sized + Committable, { fn from(commitment: &'a MessageCommitment) -> Self { - ContractSafeCommitment::from(commitment.value()) + ContractSafeCommitment(commitment.value().to_vec()) } } diff --git a/common/crypto/dkg/src/bte/mod.rs b/common/crypto/dkg/src/bte/mod.rs index 5e59a8d9d7..0fbf7602b3 100644 --- a/common/crypto/dkg/src/bte/mod.rs +++ b/common/crypto/dkg/src/bte/mod.rs @@ -302,6 +302,12 @@ pub struct Params { _h_prepared: G2Prepared, } +impl Default for Params { + fn default() -> Self { + setup() + } +} + pub fn setup() -> Params { let f0 = hash_g2(b"f0", SETUP_DOMAIN); diff --git a/common/crypto/dkg/src/lib.rs b/common/crypto/dkg/src/lib.rs index f754b16153..740dd58e22 100644 --- a/common/crypto/dkg/src/lib.rs +++ b/common/crypto/dkg/src/lib.rs @@ -3,16 +3,13 @@ // forward-secure public key encryption scheme pub mod bte; +pub mod dealing; pub mod error; pub mod interpolation; - -// this entire module is a big placeholder for whatever scheme we decide to use for the -// secure channel encryption scheme, but I would assume that the top-level API would -// remain more or less the same -pub mod dealing; pub(crate) mod share; pub(crate) mod utils; +pub use bte::{Epoch, Params}; pub use dealing::*; pub use share::*; diff --git a/contracts/Cargo.lock b/contracts/Cargo.lock index f69058b16e..1e419402b3 100644 --- a/contracts/Cargo.lock +++ b/contracts/Cargo.lock @@ -278,6 +278,8 @@ name = "contracts-common" version = "0.1.0" dependencies = [ "cosmwasm-std", + "schemars", + "serde", ] [[package]] diff --git a/contracts/coconut-dkg/src/dealers/transactions.rs b/contracts/coconut-dkg/src/dealers/transactions.rs index a772100b8e..92d996af82 100644 --- a/contracts/coconut-dkg/src/dealers/transactions.rs +++ b/contracts/coconut-dkg/src/dealers/transactions.rs @@ -5,8 +5,9 @@ use crate::constants::{INVALID_ED25519_BLACKLISTING_EXPIRATION, MINIMUM_DEPOSIT} use crate::dealers::storage as dealers_storage; use crate::ContractError; use coconut_dkg_common::types::{ - BlacklistingReason, BlockHeight, DealerDetails, EncodedBTEPublicKeyWithProof, - EncodedBTEPublicKeyWithProofRef, EncodedEd25519PublicKey, EncodedEd25519PublicKeyRef, + BlacklistingReason, BlockHeight, ContractSafeCommitment, DealerDetails, + EncodedBTEPublicKeyWithProof, EncodedBTEPublicKeyWithProofRef, EncodedEd25519PublicKey, + EncodedEd25519PublicKeyRef, }; use config::defaults::STAKE_DENOM; use cosmwasm_std::{Addr, Coin, Deps, DepsMut, Env, MessageInfo, Response}; @@ -195,8 +196,7 @@ pub fn try_commit_dealing( env: Env, info: MessageInfo, epoch_id: u32, - dealing_digest: [u8; 32], - receivers: u32, + commitment: ContractSafeCommitment, ) -> Result { todo!() } diff --git a/contracts/coconut-dkg/src/lib.rs b/contracts/coconut-dkg/src/lib.rs index 6843bdcf32..0bec70dae7 100644 --- a/contracts/coconut-dkg/src/lib.rs +++ b/contracts/coconut-dkg/src/lib.rs @@ -39,6 +39,15 @@ pub fn instantiate( return Err(ContractError::EpochStateFinishInPast); } + // if threshold was not provided in arguments, use ceil(2/3 of validators) + let system_threshold = if let Some(system_threshold) = msg.system_threshold { + system_threshold + } else { + let validators = deps.querier.query_all_validators()?.len() as u64; + // note: ceiling in integer division can be achieved via q = (x + y - 1) / y; + (2 * validators + 3 - 1) / 3 + }; + epoch_storage::CURRENT_EPOCH.save( deps.storage, &Epoch { @@ -47,6 +56,7 @@ pub fn instantiate( begun_at: env.block.height, finish_by: msg.public_key_submission_end_height, }, + system_threshold, }, )?; Ok(Response::default()) @@ -77,16 +87,8 @@ pub fn execute( ), ExecuteMsg::CommitDealing { epoch_id, - dealing_digest, - receivers, - } => dealers::transactions::try_commit_dealing( - deps, - env, - info, - epoch_id, - dealing_digest, - receivers, - ), + commitment, + } => dealers::transactions::try_commit_dealing(deps, env, info, epoch_id, commitment), ExecuteMsg::UnsafeResetAll { init_msg } => reset_contract_state(deps, env, info, init_msg), } } diff --git a/nym-wallet/Cargo.lock b/nym-wallet/Cargo.lock index b544e93531..a70b15747e 100644 --- a/nym-wallet/Cargo.lock +++ b/nym-wallet/Cargo.lock @@ -726,6 +726,8 @@ name = "contracts-common" version = "0.1.0" dependencies = [ "cosmwasm-std", + "schemars", + "serde", ] [[package]] diff --git a/validator-api/src/dkg/error.rs b/validator-api/src/dkg/error.rs index 4dce611a1a..2fe9448982 100644 --- a/validator-api/src/dkg/error.rs +++ b/validator-api/src/dkg/error.rs @@ -26,6 +26,9 @@ pub enum DkgError { #[error("Failed to recover assigned node index")] NodeIndexRecoveryError, + #[error("Failed to broadcast our message to every specified receiver ({total} receivers were specified)")] + FullBroadcastFailure { total: usize }, + #[error("todo")] MalformedDealerDetails, diff --git a/validator-api/src/dkg/events/event.rs b/validator-api/src/dkg/events/event.rs index 1c3266bc2d..dd023c100d 100644 --- a/validator-api/src/dkg/events/event.rs +++ b/validator-api/src/dkg/events/event.rs @@ -1,7 +1,7 @@ // Copyright 2022 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -use crate::dkg::networking::message::{NewDealingMessage, RemoteDealingRequestMessage}; +use crate::dkg::networking::message::NewDealingMessage; use crate::dkg::smart_contract::watcher; use std::fmt::{Display, Formatter}; diff --git a/validator-api/src/dkg/main_loop/dealing_commitment.rs b/validator-api/src/dkg/main_loop/dealing_commitment.rs index 5274f65c75..5492d788f2 100644 --- a/validator-api/src/dkg/main_loop/dealing_commitment.rs +++ b/validator-api/src/dkg/main_loop/dealing_commitment.rs @@ -2,9 +2,9 @@ // SPDX-License-Identifier: Apache-2.0 use crate::dkg::state::DkgParticipant; -use coconut_dkg_common::types::EpochId; +use coconut_dkg_common::types::{EpochId, Threshold}; use contracts_common::commitment::{Committable, DefaultHasher, Digest}; -use dkg::{Dealing, NodeIndex}; +use dkg::NodeIndex; use std::collections::BTreeMap; type ReceiversDigest = Vec; @@ -13,11 +13,28 @@ type ReceiversDigest = Vec; // note that its an ephemeral type and thus the references in here rather than owned types pub(crate) struct CommittableEpochDealing<'a> { epoch_id: EpochId, - dealing: &'a Dealing, + system_threshold: Threshold, + dealing_bytes: &'a [u8], // since all dealers are going to be using exactly the same set of receivers, - // perform commitment on a hash of receivers so that you wouldn't need to recompute the bytes every time - // you receive a dealing and verify the commitment - receivers: &'a ReceiversDigest, + // perform commitment on a hash of receivers so that we wouldn't need to recompute the bytes every time + // we receive a dealing and want to verify the commitment + receivers_digest: &'a ReceiversDigest, +} + +impl<'a> CommittableEpochDealing<'a> { + pub(crate) fn new( + epoch_id: EpochId, + system_threshold: Threshold, + dealing_bytes: &'a [u8], + receivers_digest: &'a ReceiversDigest, + ) -> Self { + CommittableEpochDealing { + epoch_id, + system_threshold, + dealing_bytes, + receivers_digest, + } + } } pub(crate) fn hash_receivers(receivers: &BTreeMap) -> ReceiversDigest { @@ -33,6 +50,11 @@ impl<'a> Committable for CommittableEpochDealing<'a> { type DigestAlgorithm = DefaultHasher; fn to_bytes(&self) -> Vec { - todo!() + let mut bytes = Vec::new(); + bytes.extend_from_slice(&self.epoch_id.to_be_bytes()); + bytes.extend_from_slice(&self.system_threshold.to_be_bytes()); + bytes.extend_from_slice(self.dealing_bytes); + bytes.extend_from_slice(self.receivers_digest); + bytes } } diff --git a/validator-api/src/dkg/main_loop/mod.rs b/validator-api/src/dkg/main_loop/mod.rs index 844c06c483..cac246a5b3 100644 --- a/validator-api/src/dkg/main_loop/mod.rs +++ b/validator-api/src/dkg/main_loop/mod.rs @@ -1,15 +1,23 @@ // Copyright 2022 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 +use crate::dkg::error::DkgError; use crate::dkg::events::DispatcherSender; +use crate::dkg::main_loop::dealing_commitment::{hash_receivers, CommittableEpochDealing}; +use crate::dkg::networking::message::OffchainDkgMessage; +use crate::dkg::networking::sender::Broadcaster; use crate::dkg::smart_contract::publisher::Publisher; use crate::dkg::smart_contract::watcher; use crate::dkg::smart_contract::watcher::{DealerChange, EventType}; -use crate::dkg::state::{DkgParticipant, DkgState, Malformation, MalformedDealer}; -use coconut_dkg_common::types::{Addr, BlockHeight, DealerDetails}; +use crate::dkg::state::{DkgParticipant, DkgState, Malformation, MalformedDealer, StateShare}; +use coconut_dkg_common::types::{Addr, BlockHeight, DealerDetails, Epoch, EpochId}; +use contracts_common::commitment::Committable; +use dkg::bte::encrypt_shares; +use dkg::{Dealing, Params}; use futures::channel::mpsc; use futures::StreamExt; use log::{debug, info, trace}; +use rand::RngCore; use std::net::SocketAddr; use validator_client::nymd::SigningCosmWasmClient; @@ -21,7 +29,10 @@ pub(crate) type ContractEventsReceiver = mpsc::UnboundedReceiver pub(crate) type ContractEventsSender = mpsc::UnboundedSender; // it is driven by events received from the contract watcher -pub(crate) struct ProcessingLoop { +pub(crate) struct ProcessingLoop { + dkg_params: dkg::Params, + + rng: R, // technically we could have used `StateAccessor` here as well, but I really want to keep // `StateAccessor` pure from anything that would have to modify `dkg_state` dkg_state: DkgState, @@ -29,32 +40,41 @@ pub(crate) struct ProcessingLoop { contract_events_receiver: ContractEventsReceiver, contract_publisher: Publisher, - // network_sender: Sender - - // TODO: this HAS to go away from here as it doesn't fit. but I've put it here temporarily to validate dealer registration - network_host: SocketAddr, } -impl ProcessingLoop +impl ProcessingLoop where C: SigningCosmWasmClient + Send + Sync, { pub(crate) fn new( + rng: R, dkg_state: DkgState, dispatcher_sender: DispatcherSender, contract_events_receiver: ContractEventsReceiver, contract_publisher: Publisher, - network_host: SocketAddr, ) -> Self { ProcessingLoop { + dkg_params: Params::default(), + rng, dkg_state, dispatcher_sender, contract_events_receiver, contract_publisher, - network_host, } } + fn random_request_id(&mut self) -> u64 { + self.rng.next_u64() + } + + async fn broadcast(&self, msg: OffchainDkgMessage, addresses: Vec) { + todo!() + } + + async fn send_to(&self, msg: OffchainDkgMessage, address: SocketAddr) { + todo!() + } + async fn raise_malformed_dealer_complaint( &self, dealer_address: Addr, @@ -67,6 +87,112 @@ where // todo!() } + async fn broadcast_dealing( + &mut self, + epoch_id: EpochId, + dealing: Dealing, + addresses: Vec, + ) -> Result<(), DkgError> { + let dealing_bytes = dealing.to_bytes(); + let signature = self.dkg_state.sign_dealing(epoch_id, &dealing_bytes).await; + let request_id = self.random_request_id(); + + let msg = + OffchainDkgMessage::new_dealing_message(request_id, epoch_id, dealing_bytes, signature); + + info!( + "broadcasting dealing for epoch {} to {} other parties", + epoch_id, + addresses.len() + ); + trace!("parties getting the dealing: {:?}", addresses); + + Broadcaster::new(addresses) + .broadcast_with_feedback(msg) + .await + } + + async fn produce_and_share_dealing(&mut self, epoch: Epoch) + where + R: RngCore, + { + let self_index = self.dkg_state.assigned_index().await; + let self_host = self.dkg_state.network_address().await; + let receivers = self.dkg_state.ordered_receivers().await; + let receivers_digest = hash_receivers(&receivers); + + let dkg_receivers = receivers + .iter() + .map(|(k, v)| (*k, *v.bte_public_key.public_key())) + .collect(); + + // make sure we don't include ourselves (also note: at this point the addresses dont have to be ordered) + let remote_hosts = receivers + .values() + .map(|receiver| receiver.remote_address) + .filter(|addr| addr != &self_host) + .collect::>(); + + let threshold = epoch.system_threshold; + let dkg_epoch = dkg::Epoch::new(epoch.id); + + // for now completely ignore the idea of resharing + let (dealing, self_share) = Dealing::create( + &mut self.rng, + &self.dkg_params, + self_index, + threshold, + dkg_epoch, + &dkg_receivers, + None, + ); + + let dealing_bytes = dealing.to_bytes(); + let committable = CommittableEpochDealing::new( + epoch.id, + epoch.system_threshold, + &dealing_bytes, + &receivers_digest, + ); + let commitment = committable.produce_commitment(); + + // first publish commitment to the chain and then broadcast it to other parties + if let Err(err) = self + .contract_publisher + .submit_dealing_commitment(epoch.id, commitment.contract_safe_commitment()) + .await + { + error!("failed to submit dealing commitment to the chain - {}", err); + // TODO: should we exit the process at this point? This seem to be a rather critical problem + // as we cannot participate in DKG without that + return; + } + + // TODO: should we be broadcasting in separate task to not block the loop for processing other messages? + if let Err(err) = self + .broadcast_dealing(epoch.id, dealing, remote_hosts) + .await + { + error!("failed to broadcast our dealing to other parties - {}", err); + // TODO: again, should we exit here? + return; + } + + // TODO: think how to handle it => we want to keep it until we receive all dealings to + // derive the shared key, but we really want to avoid storing it on disk in plain + let state_share = self_share.map(|self_share| { + let (ciphertext, _) = encrypt_shares( + &[(&self_share, self.dkg_state.public_bte_key())], + dkg_epoch, + &self.dkg_params, + &mut self.rng, + ); + StateShare::new(Some(self_share), ciphertext) + }); + + self.dkg_state.post_dealing_submission(state_share).await + } + async fn deal_with_new_dealer(&self, dealer: DkgParticipant) { if dealer.bte_public_key.verify() { self.dkg_state.try_add_new_dealer(dealer).await @@ -122,7 +248,8 @@ where let chain_address = self.contract_publisher.get_address().await; let registration = self .dkg_state - .prepare_dealer_registration(chain_address, self.network_host.to_string()); + .prepare_dealer_registration(chain_address) + .await; match self .contract_publisher @@ -130,7 +257,7 @@ where registration.identity, registration.bte_key, registration.owner_signature, - registration.listening_address, + registration.network_address, ) .await { @@ -157,17 +284,22 @@ where DealerChange::Removal { address } => self.process_dealer_removal(address).await, } } - self.dkg_state.update_last_seen_height(height).await } - async fn process_event(&self, event: watcher::Event) { + async fn process_event(&mut self, event: watcher::Event) { match event.event_type { EventType::NewKeySubmission => self.process_new_key_submission(event.height).await, EventType::DealerSetChange { changes } => { self.process_dealer_changes(changes, event.height).await } - EventType::NewDealingCommitment => todo!(), + EventType::NewDealingCommitment { epoch } => { + self.produce_and_share_dealing(epoch).await + } + EventType::NoChange => { + trace!("no change in the contract, going to only update the last seen height"); + } } + self.dkg_state.update_last_seen_height(event.height).await } pub(crate) async fn run(&mut self) { diff --git a/validator-api/src/dkg/mod.rs b/validator-api/src/dkg/mod.rs index 987561026e..aa7468c093 100644 --- a/validator-api/src/dkg/mod.rs +++ b/validator-api/src/dkg/mod.rs @@ -44,6 +44,8 @@ where C: SigningCosmWasmClient + Send + Sync + 'static, { info!("initialising dkg..."); + let rng = OsRng; + let (dispatcher_sender, dispatcher_receiver) = mpsc::unbounded(); let (contracts_events_sender, contracts_events_receiver) = mpsc::unbounded(); let (dealing_sender, dealing_receiver) = mpsc::unbounded(); @@ -51,6 +53,7 @@ where // TODO: change it to attempt to load from a file first let state = DkgState::initialise_fresh( &nyxd_client, + config.network_address, config.identity, config.decryption_key, config.public_key, @@ -72,14 +75,14 @@ where config.contract_polling_interval, ); let publisher = smart_contract::Publisher::new(nyxd_client); - let mut net_listener = Listener::new(config.network_address.clone(), state_accessor); + let mut net_listener = Listener::new(config.network_address, state_accessor); let mut processing_loop = ProcessingLoop::new( + rng, state, dispatcher_sender, contracts_events_receiver, publisher, - config.network_address, ); tokio::spawn(async move { event_dispatcher.run().await }); diff --git a/validator-api/src/dkg/networking/codec.rs b/validator-api/src/dkg/networking/codec.rs index 62ff1ea550..baee49cd93 100644 --- a/validator-api/src/dkg/networking/codec.rs +++ b/validator-api/src/dkg/networking/codec.rs @@ -12,10 +12,10 @@ const MAX_ALLOWED_MESSAGE_LEN: usize = 2 * 1024 * 1024; #[derive(Debug)] pub struct DkgCodec; -impl Encoder for DkgCodec { +impl<'a> Encoder<&'a OffchainDkgMessage> for DkgCodec { type Error = DkgError; - fn encode(&mut self, item: OffchainDkgMessage, dst: &mut BytesMut) -> Result<(), Self::Error> { + fn encode(&mut self, item: &OffchainDkgMessage, dst: &mut BytesMut) -> Result<(), Self::Error> { item.encode(dst) } } diff --git a/validator-api/src/dkg/networking/message.rs b/validator-api/src/dkg/networking/message.rs index 4b42006adb..2b75e9ee95 100644 --- a/validator-api/src/dkg/networking/message.rs +++ b/validator-api/src/dkg/networking/message.rs @@ -5,6 +5,7 @@ use crate::dkg::error::DkgError; use crate::dkg::networking::PROTOCOL_VERSION; use crate::dkg::state::ReceivedDealing; use bytes::{BufMut, BytesMut}; +use coconut_dkg_common::types::EpochId; use crypto::asymmetric::identity; use serde::{Deserialize, Serialize}; use std::fmt::{Display, Formatter}; @@ -13,7 +14,7 @@ use std::net::SocketAddr; use std::time::Duration; use thiserror::Error; -#[derive(Debug, Serialize, Deserialize)] +#[derive(Debug, Clone, Serialize, Deserialize)] pub enum OffchainDkgMessage { NewDealing { id: u64, @@ -33,7 +34,27 @@ pub enum OffchainDkgMessage { }, } -#[derive(Debug, Serialize, Deserialize)] +impl Display for OffchainDkgMessage { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + write!(f, "OffchainDkgMessage ")?; + match self { + OffchainDkgMessage::NewDealing { id, message } => { + write!(f, "with id {} and message: {}", id, message) + } + OffchainDkgMessage::RemoteDealingRequest { id, message } => { + write!(f, "with id {} and message: {}", id, message) + } + OffchainDkgMessage::RemoteDealingResponse { id, message } => { + write!(f, "with id {} and message: {}", id, message) + } + OffchainDkgMessage::ErrorResponse { id, message } => { + write!(f, "with id {:?} and message: {}", id, message) + } + } + } +} + +#[derive(Debug, Clone, Serialize, Deserialize)] pub struct NewDealingMessage { pub epoch_id: u32, // we keep the dealing in its serialized state as that's what is being signed (and hashed) @@ -53,19 +74,39 @@ impl Display for NewDealingMessage { } } -#[derive(Debug, Serialize, Deserialize)] +#[derive(Debug, Clone, Serialize, Deserialize)] pub struct RemoteDealingRequestMessage { pub epoch_id: u32, pub dealer: identity::PublicKey, } -#[derive(Debug, Serialize, Deserialize)] +impl Display for RemoteDealingRequestMessage { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + write!( + f, + "RemoteDealingRequestMessage for epoch {} for dealer {}", + self.epoch_id, self.dealer + ) + } +} + +#[derive(Debug, Clone, Serialize, Deserialize)] pub enum RemoteDealingResponseMessage { Available { dealing: ReceivedDealing }, Unavailable, } -#[derive(Debug, Serialize, Deserialize, Error)] +impl Display for RemoteDealingResponseMessage { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + write!(f, "RemoteDealingResponseMessage - ")?; + match self { + RemoteDealingResponseMessage::Available { .. } => write!(f, "Available"), + RemoteDealingResponseMessage::Unavailable => write!(f, "Unavailable"), + } + } +} + +#[derive(Debug, Clone, Serialize, Deserialize, Error)] pub enum ErrorResponseMessage { #[error("Received request for epoch: {requested}, while the current epoch is {current}")] InvalidEpoch { current: u32, requested: u32 }, @@ -84,17 +125,27 @@ pub enum ErrorResponseMessage { } impl OffchainDkgMessage { - pub(crate) fn new_error_response( - id: Option, - message: ErrorResponseMessage, - ) -> OffchainDkgMessage { + pub(crate) fn new_error_response(id: Option, message: ErrorResponseMessage) -> Self { OffchainDkgMessage::ErrorResponse { id, message } } - pub(crate) fn new_remote_dealing_response( + pub(crate) fn new_dealing_message( id: u64, - dealing: Option, - ) -> OffchainDkgMessage { + epoch_id: EpochId, + dealing_bytes: Vec, + dealer_signature: identity::Signature, + ) -> Self { + OffchainDkgMessage::NewDealing { + id, + message: NewDealingMessage { + epoch_id, + dealing_bytes, + dealer_signature, + }, + } + } + + pub(crate) fn new_remote_dealing_response(id: u64, dealing: Option) -> Self { let message = match dealing { Some(dealing) => RemoteDealingResponseMessage::Available { dealing }, None => RemoteDealingResponseMessage::Unavailable, @@ -111,7 +162,7 @@ impl OffchainDkgMessage { Ok(bincode::serialize(&self)?) } - fn frame(self) -> Result { + fn frame(&self) -> Result { let payload = self.try_to_bytes()?; Ok(FramedOffchainDkgMessage { header: Header { @@ -122,7 +173,7 @@ impl OffchainDkgMessage { }) } - pub(crate) fn encode(self, dst: &mut BytesMut) -> Result<(), DkgError> { + pub(crate) fn encode(&self, dst: &mut BytesMut) -> Result<(), DkgError> { dst.put(self.frame()?.into_bytes().as_ref()); Ok(()) } diff --git a/validator-api/src/dkg/networking/receiver/handler.rs b/validator-api/src/dkg/networking/receiver/handler.rs index 62bcb8d672..2306b4a98d 100644 --- a/validator-api/src/dkg/networking/receiver/handler.rs +++ b/validator-api/src/dkg/networking/receiver/handler.rs @@ -36,7 +36,7 @@ impl ConnectionHandler { } async fn send_response(&mut self, response_message: OffchainDkgMessage) { - if let Err(err) = self.conn.send(response_message).await { + if let Err(err) = self.conn.send(&response_message).await { warn!("Failed to send response back to {} - {}", self.remote, err) } } diff --git a/validator-api/src/dkg/networking/sender/mod.rs b/validator-api/src/dkg/networking/sender/mod.rs index 87d5c392c8..1f78c8fdae 100644 --- a/validator-api/src/dkg/networking/sender/mod.rs +++ b/validator-api/src/dkg/networking/sender/mod.rs @@ -1,2 +1,272 @@ // Copyright 2022 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 + +use crate::dkg::error::DkgError; +use crate::dkg::networking::codec::DkgCodec; +use crate::dkg::networking::message::OffchainDkgMessage; +use futures::channel::mpsc; +use futures::{stream, SinkExt, StreamExt}; +use log::{debug, info, trace, warn}; +use std::io; +use std::net::SocketAddr; +use std::time::Duration; +use tokio::net::TcpStream; +use tokio::time::timeout; +use tokio_util::codec::Framed; + +// TODO: for now just leave them here and make it configurable with proper config later +const DEFAULT_CONCURRENCY: usize = 5; +const DEFAULT_CONNECTION_TIMEOUT: Duration = Duration::from_secs(10); +const DEFAULT_RESPONSE_TIMEOUT: Duration = Duration::from_secs(60); +const DEFAULT_SEND_TIMEOUT: Duration = Duration::from_secs(10); + +pub(crate) struct SendResponse { + source: SocketAddr, + response: Result, DkgError>, +} + +type FeedbackSender = mpsc::UnboundedSender; + +pub(crate) struct Broadcaster { + addresses: Vec, + concurrency_level: usize, + connection_timeout: Duration, + response_timeout: Duration, + send_timeout: Duration, +} + +impl Broadcaster { + pub(crate) fn new(addresses: Vec) -> Self { + Broadcaster { + addresses, + concurrency_level: DEFAULT_CONCURRENCY, + connection_timeout: DEFAULT_CONNECTION_TIMEOUT, + response_timeout: DEFAULT_RESPONSE_TIMEOUT, + send_timeout: DEFAULT_SEND_TIMEOUT, + } + } + + pub(crate) fn with_concurrency_level(mut self, concurrency_level: usize) -> Self { + self.concurrency_level = concurrency_level; + self + } + + pub(crate) fn with_connection_timeout(mut self, connection_timeout: Duration) -> Self { + self.connection_timeout = connection_timeout; + self + } + + pub(crate) fn with_response_timeout(mut self, response_timeout: Duration) -> Self { + self.response_timeout = response_timeout; + self + } + + pub(crate) fn with_send_timeout(mut self, send_timeout: Duration) -> Self { + self.send_timeout = send_timeout; + self + } + + pub(crate) fn set_addresses(&mut self, new_addresses: Vec) { + self.addresses = new_addresses; + } + + fn create_broadcast_configs( + &self, + message: OffchainDkgMessage, + feedback_sender: Option, + ) -> Vec { + self.addresses + .iter() + .map(|&address| BroadcastConfig { + address, + connection_timeout: self.connection_timeout, + response_timeout: Some(self.response_timeout), + send_timeout: self.send_timeout, + feedback_sender: feedback_sender.clone(), + message: message.clone(), + }) + .collect() + } + + pub(crate) async fn broadcast_with_feedback( + &self, + msg: OffchainDkgMessage, + ) -> Result<(), DkgError> { + if self.addresses.is_empty() { + warn!("attempting to broadcast {} while no remotes are known", msg); + return Ok(()); + } + + 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 failures = 0; + + 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)) => { + match response.response { + Err(err) => { + failures += 1; + warn!("we failed to broadcast to {} - {}", response.source, err) + } + Ok(Some(res)) => { + // TODO: figure out what to do with the replies exactly in the broadcast case... + if let OffchainDkgMessage::ErrorResponse { message, .. } = res { + warn!( + "we received an error response from {} - {}", + response.source, message + ) + } else { + info!( + "{} provided a non-error response to our broadcast! - {}", + response.source, res + ) + } + } + // the expected case + Ok(None) => debug!("{} didn't provide any reply", response.source), + } + } + 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))); + + // if we failed to send to everyone, return an error, otherwise, assuming at least a single + // receiver got our dealing, it can be gossiped through (assuming the problem was on our side, + // i.e. the other receivers will actually want to receive the dealing) + if failures == self.addresses.len() { + error!("we failed to broadcast to every single receiver"); + return Err(DkgError::FullBroadcastFailure { total: failures }); + } else { + Ok(()) + } + } + + #[allow(dead_code)] + pub(crate) async fn broadcast(&self, msg: OffchainDkgMessage) { + 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, + connection_timeout: Duration, + response_timeout: Option, + send_timeout: Duration, + feedback_sender: Option, + message: OffchainDkgMessage, +} + +impl BroadcastConfig { + async fn send(self) { + let response = send_message( + self.address, + &self.message, + self.connection_timeout, + self.send_timeout, + self.response_timeout, + ) + .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(SendResponse { + source: self.address, + response, + }); + } else if let Err(err) = 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 + ) + } + } +} + +// this connection only exists for a single message +pub(crate) struct EphemeralConnection { + 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, DkgCodec); + Ok(Self { conn: framed_conn }) + } + + pub(crate) fn remote(&self) -> io::Result { + self.conn.get_ref().peer_addr() + } + + pub(crate) async fn send( + &mut self, + message: &OffchainDkgMessage, + send_timeout: Duration, + response_timeout: Option, + ) -> Result, DkgError> { + trace!("attempting to send to {}", self.remote()?); + match timeout(send_timeout, self.conn.send(message)).await { + Err(_timeout) => { + return Err(DkgError::Networking(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 send_message( + address: SocketAddr, + message: &OffchainDkgMessage, + connection_timeout: Duration, + send_timeout: Duration, + response_timeout: Option, +) -> Result, DkgError> { + let mut conn = EphemeralConnection::connect(address, connection_timeout).await?; + conn.send(message, send_timeout, response_timeout).await +} diff --git a/validator-api/src/dkg/smart_contract/publisher/mod.rs b/validator-api/src/dkg/smart_contract/publisher/mod.rs index 07f9389409..e719a97a46 100644 --- a/validator-api/src/dkg/smart_contract/publisher/mod.rs +++ b/validator-api/src/dkg/smart_contract/publisher/mod.rs @@ -3,7 +3,10 @@ use crate::dkg::error::DkgError; use crate::Client; -use coconut_dkg_common::types::{EncodedBTEPublicKeyWithProof, EncodedEd25519PublicKey, NodeIndex}; +use coconut_dkg_common::types::{ + EncodedBTEPublicKeyWithProof, EncodedEd25519PublicKey, EpochId, NodeIndex, +}; +use contracts_common::commitment::ContractSafeCommitment; use validator_client::nymd::{AccountId, SigningCosmWasmClient}; pub(crate) struct Publisher { @@ -45,8 +48,18 @@ where Err(DkgError::NodeIndexRecoveryError) } - pub(crate) async fn submit_dealing_commitment(&self) { - // self.client.submit_dealing_commitment().await; - // + pub(crate) async fn submit_dealing_commitment( + &self, + epoch_id: EpochId, + commitment: ContractSafeCommitment, + ) -> Result<(), DkgError> { + info!( + "submitting dealing commitment for epoch {} to the chain. Commitment is: {}", + epoch_id, commitment + ); + self.client + .submit_dealing_commitment(epoch_id, commitment) + .await?; + Ok(()) } } diff --git a/validator-api/src/dkg/smart_contract/watcher/event.rs b/validator-api/src/dkg/smart_contract/watcher/event.rs index 91115ff875..1638bf6e39 100644 --- a/validator-api/src/dkg/smart_contract/watcher/event.rs +++ b/validator-api/src/dkg/smart_contract/watcher/event.rs @@ -1,7 +1,7 @@ // Copyright 2022 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -use coconut_dkg_common::types::{Addr, BlockHeight, DealerDetails}; +use coconut_dkg_common::types::{Addr, BlockHeight, DealerDetails, Epoch}; use std::fmt::{Display, Formatter}; #[derive(Debug)] @@ -34,9 +34,10 @@ pub(crate) enum DealerChange { #[derive(Debug)] pub(crate) enum EventType { + NoChange, NewKeySubmission, DealerSetChange { changes: Vec }, - NewDealingCommitment, + NewDealingCommitment { epoch: Epoch }, } impl Display for EventType { @@ -47,7 +48,10 @@ impl Display for EventType { EventType::DealerSetChange { changes } => { write!(f, "DealerSetChange with {} changes", changes.len()) } - EventType::NewDealingCommitment => write!(f, "NewDealingCommitment"), + EventType::NewDealingCommitment { epoch } => { + write!(f, "NewDealingCommitment for epoch {}", epoch.id) + } + EventType::NoChange => write!(f, "NoChange"), } } } diff --git a/validator-api/src/dkg/smart_contract/watcher/mod.rs b/validator-api/src/dkg/smart_contract/watcher/mod.rs index 59a72e47b0..d282ef7717 100644 --- a/validator-api/src/dkg/smart_contract/watcher/mod.rs +++ b/validator-api/src/dkg/smart_contract/watcher/mod.rs @@ -157,21 +157,64 @@ where .await } + // async fn check_for_dealing_commitment( + // &self, + // contract_dealers: &HashMap, + // current_height: BlockHeight, + // ) -> Result<(), DkgError> { + // // note: normally we want to avoid the unchecked API, however, in this case it's fine as the + // // `AccountId` is coming from the client is valid as it has been derived directly from the provided mnemonic, + // // and hence we are certain it is valid + // let address = Addr::unchecked(self.client.address().await.as_ref()); + // if !contract_dealers.contains_key(&address) { + // // our key is not present in contract dealers, check if we think we have submitted it + // if !self.state_accessor.has_submitted_keys().await { + // // if we just transitioned into `PublicKeySubmission` and we haven't submitted our own keys + // // we should emit event to do just that + // debug!("we never registered our own dkg keys"); + // self.state_accessor + // .push_new_key_submission_event(current_height) + // .await; + // } else { + // // check if we got blacklisted, since we think we have submitted our own key... + // let blacklisting = self.client.get_blacklisting(address.into_string()).await?; + // + // if blacklisting.is_blacklisted(current_height) { + // warn!("our dealer is blacklisted - {}. We cannot participate in this round of DKG", blacklisting.unchecked_get_blacklisting()); + // // TODO: what to do about it? can we do anything about it? + // } else { + // // we've been blacklisted in the past, but it has already expired + // debug!( + // "our dealer has been blacklisted in the past, but it has already expired" + // ); + // self.state_accessor + // .push_new_key_submission_event(current_height) + // .await; + // } + // } + // } else { + // // TODO: change to trace + // debug!("our dkg key is already registered in the dkg contract") + // } + // + // Ok(()) + // } + async fn dealing_exchange_actions(&self) -> Result<(), DkgError> { let current_height = self.client.current_block_height().await?.value(); - // let contract_dealers = self - // .client - // .get_current_dealers() - // .await? - // .into_iter() - // .map(|dealer| (dealer.address.clone(), dealer)) - // .collect::>(); + let contract_dealers = self + .client + .get_current_dealers() + .await? + .into_iter() + .map(|dealer| (dealer.address.clone(), dealer)) + .collect::>(); // // self.check_for_own_submission(&contract_dealers, current_height) // .await?; // self.check_for_dealers(contract_dealers, current_height) // .await - + todo!() } diff --git a/validator-api/src/dkg/state/mod.rs b/validator-api/src/dkg/state/mod.rs index 3d29dbedec..a4096bdbcd 100644 --- a/validator-api/src/dkg/state/mod.rs +++ b/validator-api/src/dkg/state/mod.rs @@ -1,27 +1,27 @@ // Copyright 2022 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 +use crate::dkg::error::DkgError; use crate::Client; use coconut_dkg_common::types::{ Addr, BlockHeight, DealerDetails, EncodedBTEPublicKeyWithProof, EncodedEd25519PublicKey, Epoch, - NodeIndex, + EpochId, NodeIndex, }; use crypto::asymmetric::identity; -use dkg::{bte, Dealing}; +use dkg::bte::Ciphertexts; +use dkg::{bte, Dealing, Share}; use futures::lock::Mutex; use log::debug; use log::error; use serde::{Deserialize, Serialize}; -use std::collections::HashMap; -use std::hash::Hash; +use std::collections::{BTreeMap, HashMap}; use std::net::SocketAddr; use std::sync::Arc; +use validator_client::nymd::{AccountId, SigningCosmWasmClient}; mod accessor; -use crate::dkg::error::DkgError; pub(crate) use accessor::StateAccessor; -use validator_client::nymd::{AccountId, SigningCosmWasmClient}; type IdentityBytes = [u8; identity::PUBLIC_KEY_LENGTH]; @@ -116,7 +116,7 @@ pub(crate) struct DealerRegistration { pub(crate) identity: EncodedEd25519PublicKey, pub(crate) bte_key: EncodedBTEPublicKeyWithProof, pub(crate) owner_signature: String, - pub(crate) listening_address: String, + pub(crate) network_address: String, } #[derive(Debug, Clone)] @@ -133,20 +133,55 @@ struct Keys { bte_public_key: bte::PublicKeyWithProof, } +fn empty_ciphertexts() -> Ciphertexts { + // this is super temporary and never meant to be used in the long run + Ciphertexts { + rr: Default::default(), + ss: Default::default(), + zz: Default::default(), + ciphertext_chunks: vec![], + } +} + +// we want to avoid every storing the actual share in plain on the disk, so only the encrypted version +// is potentially stored and the actual share is decrypted when actually needed (i.e. during aggregation) +// however, if we don't have to serialize data to the disk and everything is kept in memory, there's no +// point in not using what we already have +#[derive(Debug, Serialize, Deserialize)] +pub(crate) struct StateShare { + #[serde(skip)] + share: Option, + + // the reason for the skip is that I want the code to compile and I haven't yet implemented serde for Ciphertexts : ) + #[serde(skip, default = "empty_ciphertexts")] + encrypted_share: Ciphertexts, +} + +impl StateShare { + pub(crate) fn new(share: Option, encrypted_share: Ciphertexts) -> Self { + StateShare { + share, + encrypted_share, + } + } +} + #[derive(Debug, Serialize, Deserialize)] struct DkgStateInner { submitted_keys: bool, submitted_commitment: bool, submitted_verification_keys: bool, + network_address: SocketAddr, assigned_index: NodeIndex, last_seen_height: BlockHeight, - current_epoch: Epoch, expected_epoch_dealing_digests: HashMap, - // we need to keep track of all bad dealers as well so that we wouldn't attempt to complaint about them + self_share: Option, + + // we need to keep track of all bad dealers as well so that we wouldn't attempt to complain about them // repeatedly bad_dealers: HashMap, current_epoch_dealers: HashMap, @@ -158,6 +193,7 @@ impl DkgState { // this should only ever be called once, during init pub(crate) async fn initialise_fresh( nyxd_client: &Client, + network_address: SocketAddr, identity: identity::KeyPair, bte_decryption_key: bte::DecryptionKey, bte_public_key: bte::PublicKeyWithProof, @@ -177,10 +213,12 @@ impl DkgState { submitted_keys: false, submitted_commitment: false, submitted_verification_keys: false, + network_address, assigned_index: 0, last_seen_height: 0, current_epoch, expected_epoch_dealing_digests: HashMap::new(), + self_share: None, bad_dealers: HashMap::new(), current_epoch_dealers: HashMap::new(), verified_epoch_dealings: HashMap::new(), @@ -208,12 +246,36 @@ impl DkgState { // } + pub(crate) async fn ordered_receivers(&self) -> BTreeMap { + self.inner_state + .lock() + .await + .current_epoch_dealers + .values() + .map(|participant| (participant.node_index, participant.clone())) + .collect() + } + + pub(crate) async fn network_address(&self) -> SocketAddr { + self.inner_state.lock().await.network_address + } + + pub(crate) async fn assigned_index(&self) -> NodeIndex { + self.inner_state.lock().await.assigned_index + } + pub(crate) async fn post_key_submission(&self, assigned_index: NodeIndex) { let mut guard = self.inner_state.lock().await; guard.submitted_keys = true; guard.assigned_index = assigned_index; } + pub(crate) async fn post_dealing_submission(&self, share: Option) { + let mut guard = self.inner_state.lock().await; + guard.submitted_commitment = true; + guard.self_share = share; + } + pub(crate) async fn is_dealers_remote_address(&self, remote: SocketAddr) -> (bool, Epoch) { let guard = self.inner_state.lock().await; let epoch = guard.current_epoch; @@ -318,30 +380,44 @@ impl DkgState { } } - pub(crate) fn prepare_dealer_registration( + pub(crate) fn sign(&self, message: &[u8]) -> identity::Signature { + self.keys.identity.private_key().sign(message) + } + + pub(crate) async fn prepare_dealer_registration( &self, chain_address: AccountId, - listening_address: String, ) -> DealerRegistration { let bte_key = bs58::encode(&self.keys.bte_public_key.to_bytes()).into_string(); + let network_address = self.network_address().await.to_string(); // chain_address || host || bte_keys let mut plaintext = chain_address.to_string(); - plaintext.push_str(&listening_address); + plaintext.push_str(&network_address); plaintext.push_str(&bte_key); - let owner_signature = self - .keys - .identity - .private_key() - .sign(plaintext.as_bytes()) - .to_base58_string(); + let owner_signature = self.sign(plaintext.as_bytes()).to_base58_string(); DealerRegistration { identity: self.keys.identity.public_key().to_base58_string(), bte_key, owner_signature, - listening_address, + network_address, } } + + pub(crate) fn public_bte_key(&self) -> &bte::PublicKey { + self.keys.bte_public_key.public_key() + } + + pub(crate) async fn sign_dealing( + &self, + epoch_id: EpochId, + dealing_bytes: &[u8], + ) -> identity::Signature { + let mut bytes = Vec::with_capacity(dealing_bytes.len() + std::mem::size_of::()); + bytes.extend_from_slice(&epoch_id.to_be_bytes()); + bytes.extend_from_slice(dealing_bytes); + self.sign(&bytes) + } }