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

Validator API-side logic for dealing broadcast

Browse Source
This commit is contained in:
Jędrzej Stuczyński
2022-05-19 15:11:41 +01:00
parent 487a3a6cd8
commit 34b1fc8989
24 changed files with 772 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+2
View File
@@ -735,6 +735,8 @@ dependencies = [
"blake3",
"cosmwasm-std",
"digest 0.10.3",
"schemars",
"serde",
]
[[package]]
common/cosmwasm-smart-contracts/coconut-dkg-contract/src/msg.rs
+2 -1
View File
@@ -1,7 +1,7 @@
// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
// 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<Threshold>,
}
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, JsonSchema)]
common/cosmwasm-smart-contracts/coconut-dkg-contract/src/types.rs
+10 -1
View File
@@ -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,
})
}
}
common/cosmwasm-smart-contracts/contracts-common/Cargo.toml
+4 -1
View File
@@ -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]
common/cosmwasm-smart-contracts/contracts-common/src/commitment.rs
+33 -2
View File
@@ -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<u8>;
#[derive(
Debug, Clone, Ord, PartialOrd, Eq, PartialEq, Hash, Serialize, Deserialize, JsonSchema,
)]
pub struct ContractSafeCommitment(Vec<u8>);
impl Deref for ContractSafeCommitment {
type Target = Vec<u8>;
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<T>) -> Self {
ContractSafeCommitment::from(commitment.value())
ContractSafeCommitment(commitment.value().to_vec())
}
}
common/crypto/dkg/src/bte/mod.rs
+6
View File
@@ -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);
common/crypto/dkg/src/lib.rs
+2 -5
View File
@@ -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::*;
contracts/Cargo.lock
Generated
+2
View File
@@ -278,6 +278,8 @@ name = "contracts-common"
version = "0.1.0"
dependencies = [
"cosmwasm-std",
"schemars",
"serde",
]
[[package]]
contracts/coconut-dkg/src/dealers/transactions.rs
+4 -4
View File
@@ -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<Response, ContractError> {
todo!()
}
contracts/coconut-dkg/src/lib.rs
+12 -10
View File
@@ -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),
}
}
nym-wallet/Cargo.lock
Generated
+2
View File
@@ -726,6 +726,8 @@ name = "contracts-common"
version = "0.1.0"
dependencies = [
"cosmwasm-std",
"schemars",
"serde",
]
[[package]]
validator-api/src/dkg/error.rs
+3
View File
@@ -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,
validator-api/src/dkg/events/event.rs
+1 -1
View File
@@ -1,7 +1,7 @@
// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
// 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};
validator-api/src/dkg/main_loop/dealing_commitment.rs
+29 -7
View File
@@ -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<u8>;
@@ -13,11 +13,28 @@ type ReceiversDigest = Vec<u8>;
// 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<NodeIndex, DkgParticipant>) -> ReceiversDigest {
@@ -33,6 +50,11 @@ impl<'a> Committable for CommittableEpochDealing<'a> {
type DigestAlgorithm = DefaultHasher;
fn to_bytes(&self) -> Vec<u8> {
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
}
}
validator-api/src/dkg/main_loop/mod.rs
+147 -15
View File
@@ -1,15 +1,23 @@
// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
// 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<watcher::Event>
pub(crate) type ContractEventsSender = mpsc::UnboundedSender<watcher::Event>;
// it is driven by events received from the contract watcher
pub(crate) struct ProcessingLoop<C> {
pub(crate) struct ProcessingLoop<C, R> {
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<C> {
contract_events_receiver: ContractEventsReceiver,
contract_publisher: Publisher<C>,
// 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<C> ProcessingLoop<C>
impl<C, R: RngCore> ProcessingLoop<C, R>
where
C: SigningCosmWasmClient + Send + Sync,
{
pub(crate) fn new(
rng: R,
dkg_state: DkgState,
dispatcher_sender: DispatcherSender,
contract_events_receiver: ContractEventsReceiver,
contract_publisher: Publisher<C>,
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<SocketAddr>) {
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<SocketAddr>,
) -> 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::<Vec<_>>();
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) {
validator-api/src/dkg/mod.rs
+5 -2
View File
@@ -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 });
validator-api/src/dkg/networking/codec.rs
+2 -2
View File
@@ -12,10 +12,10 @@ const MAX_ALLOWED_MESSAGE_LEN: usize = 2 * 1024 * 1024;
#[derive(Debug)]
pub struct DkgCodec;
impl Encoder<OffchainDkgMessage> 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)
}
}
validator-api/src/dkg/networking/message.rs
+65 -14
View File
@@ -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<u64>,
message: ErrorResponseMessage,
) -> OffchainDkgMessage {
pub(crate) fn new_error_response(id: Option<u64>, message: ErrorResponseMessage) -> Self {
OffchainDkgMessage::ErrorResponse { id, message }
}
pub(crate) fn new_remote_dealing_response(
pub(crate) fn new_dealing_message(
id: u64,
dealing: Option<ReceivedDealing>,
) -> OffchainDkgMessage {
epoch_id: EpochId,
dealing_bytes: Vec<u8>,
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<ReceivedDealing>) -> 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<FramedOffchainDkgMessage, DkgError> {
fn frame(&self) -> Result<FramedOffchainDkgMessage, DkgError> {
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(())
}
validator-api/src/dkg/networking/receiver/handler.rs
+1 -1
View File
@@ -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)
}
}
validator-api/src/dkg/networking/sender/mod.rs
+270
View File
@@ -1,2 +1,272 @@
// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
// 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<Option<OffchainDkgMessage>, DkgError>,
}
type FeedbackSender = mpsc::UnboundedSender<SendResponse>;
pub(crate) struct Broadcaster {
addresses: Vec<SocketAddr>,
concurrency_level: usize,
connection_timeout: Duration,
response_timeout: Duration,
send_timeout: Duration,
}
impl Broadcaster {
pub(crate) fn new(addresses: Vec<SocketAddr>) -> 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<SocketAddr>) {
self.addresses = new_addresses;
}
fn create_broadcast_configs(
&self,
message: OffchainDkgMessage,
feedback_sender: Option<FeedbackSender>,
) -> Vec<BroadcastConfig> {
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<Duration>,
send_timeout: Duration,
feedback_sender: Option<FeedbackSender>,
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<TcpStream, DkgCodec>,
}
impl EphemeralConnection {
pub(crate) async fn connect(
address: SocketAddr,
connection_timeout: Duration,
) -> io::Result<Self> {
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<SocketAddr> {
self.conn.get_ref().peer_addr()
}
pub(crate) async fn send(
&mut self,
message: &OffchainDkgMessage,
send_timeout: Duration,
response_timeout: Option<Duration>,
) -> Result<Option<OffchainDkgMessage>, 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<Duration>,
) -> Result<Option<OffchainDkgMessage>, DkgError> {
let mut conn = EphemeralConnection::connect(address, connection_timeout).await?;
conn.send(message, send_timeout, response_timeout).await
}
validator-api/src/dkg/smart_contract/publisher/mod.rs
+17 -4
View File
@@ -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<C> {
@@ -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(())
}
}
validator-api/src/dkg/smart_contract/watcher/event.rs
+7 -3
View File
@@ -1,7 +1,7 @@
// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
// 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<DealerChange> },
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"),
}
}
}
validator-api/src/dkg/smart_contract/watcher/mod.rs
+51 -8
View File
@@ -157,21 +157,64 @@ where
.await
}
// async fn check_for_dealing_commitment(
// &self,
// contract_dealers: &HashMap<Addr, DealerDetails>,
// 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::<HashMap<_, _>>();
let contract_dealers = self
.client
.get_current_dealers()
.await?
.into_iter()
.map(|dealer| (dealer.address.clone(), dealer))
.collect::<HashMap<_, _>>();
//
// self.check_for_own_submission(&contract_dealers, current_height)
// .await?;
// self.check_for_dealers(contract_dealers, current_height)
// .await
todo!()
}
validator-api/src/dkg/state/mod.rs
+95 -19
View File
@@ -1,27 +1,27 @@
// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
// 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<Share>,
// 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<Share>, 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<IdentityBytes, [u8; 32]>,
// we need to keep track of all bad dealers as well so that we wouldn't attempt to complaint about them
self_share: Option<StateShare>,
// 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<Addr, MalformedDealer>,
current_epoch_dealers: HashMap<IdentityBytes, DkgParticipant>,
@@ -158,6 +193,7 @@ impl DkgState {
// this should only ever be called once, during init
pub(crate) async fn initialise_fresh<C>(
nyxd_client: &Client<C>,
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<NodeIndex, DkgParticipant> {
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<StateShare>) {
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::<EpochId>());
bytes.extend_from_slice(&epoch_id.to_be_bytes());
bytes.extend_from_slice(dealing_bytes);
self.sign(&bytes)
}
}