From 8bc12bc079a5cc295667d61e4f07f03febd290d9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Fri, 13 May 2022 16:20:34 +0100 Subject: [PATCH] full event propagation and execution of dkg key submission --- .../validator-client/src/client.rs | 2 +- .../coconut-dkg-contract/src/dealer.rs | 17 +++- common/crypto/src/asymmetric/identity/mod.rs | 1 + validator-api/src/dkg/dealing_processing.rs | 4 +- validator-api/src/dkg/events/dispatcher.rs | 4 +- validator-api/src/dkg/main_loop/mod.rs | 8 ++ validator-api/src/dkg/mod.rs | 8 +- .../src/dkg/networking/receiver/listener.rs | 3 + .../src/dkg/smart_contract/watcher/event.rs | 2 + .../src/dkg/smart_contract/watcher/mod.rs | 81 ++++++++++++--- validator-api/src/dkg/state/accessor.rs | 18 +++- validator-api/src/dkg/state/mod.rs | 98 ++++++++++++++----- validator-api/src/nymd_client.rs | 26 ++++- 13 files changed, 218 insertions(+), 54 deletions(-) diff --git a/common/client-libs/validator-client/src/client.rs b/common/client-libs/validator-client/src/client.rs index abc6284125..8ba3db4d2c 100644 --- a/common/client-libs/validator-client/src/client.rs +++ b/common/client-libs/validator-client/src/client.rs @@ -24,7 +24,7 @@ use crate::nymd::{ #[cfg(feature = "dkg")] use crate::nymd::{traits::DkgClient, SigningCosmWasmClient}; #[cfg(feature = "dkg")] -use coconut_dkg_common::types::{DealerDetails, BlacklistedDealer}; +use coconut_dkg_common::types::{BlacklistedDealer, DealerDetails}; #[cfg(feature = "nymd-client")] use mixnet_contract_common::{ mixnode::DelegationEvent, ContractStateParams, Delegation, IdentityKey, Interval, diff --git a/common/cosmwasm-smart-contracts/coconut-dkg-contract/src/dealer.rs b/common/cosmwasm-smart-contracts/coconut-dkg-contract/src/dealer.rs index dec1002f86..05863335cf 100644 --- a/common/cosmwasm-smart-contracts/coconut-dkg-contract/src/dealer.rs +++ b/common/cosmwasm-smart-contracts/coconut-dkg-contract/src/dealer.rs @@ -42,13 +42,13 @@ impl Display for Blacklisting { if let Some(expiration) = self.expiration { write!( f, - "blacklisted at block height {}. reason given: {}. Expires at: {}", + "blacklisted at block height {}. reason given: {}. expires at: {}", self.height, self.height, expiration ) } else { write!( f, - "blacklisted at block height {}. reason given: {}", + "permanently blacklisted at block height {}. reason given: {}", self.height, self.height ) } @@ -162,4 +162,17 @@ impl BlacklistingResponse { blacklisting, } } + + pub fn is_blacklisted(&self, current_height: BlockHeight) -> bool { + match self.blacklisting { + None => false, + Some(blacklisting) => !blacklisting.has_expired(current_height), + } + } + + pub fn unchecked_get_blacklisting(&self) -> &Blacklisting { + self.blacklisting + .as_ref() + .expect("dealer is not blacklisted") + } } diff --git a/common/crypto/src/asymmetric/identity/mod.rs b/common/crypto/src/asymmetric/identity/mod.rs index 2a03d50042..39927e1325 100644 --- a/common/crypto/src/asymmetric/identity/mod.rs +++ b/common/crypto/src/asymmetric/identity/mod.rs @@ -48,6 +48,7 @@ impl fmt::Display for Ed25519RecoveryError { impl std::error::Error for Ed25519RecoveryError {} /// Keypair for usage in ed25519 EdDSA. +#[derive(Debug)] pub struct KeyPair { private_key: PrivateKey, public_key: PublicKey, diff --git a/validator-api/src/dkg/dealing_processing.rs b/validator-api/src/dkg/dealing_processing.rs index c13db51eec..18ba979438 100644 --- a/validator-api/src/dkg/dealing_processing.rs +++ b/validator-api/src/dkg/dealing_processing.rs @@ -5,7 +5,7 @@ use crate::dkg::networking::message::NewDealingMessage; use crate::dkg::state::DkgState; use futures::channel::mpsc; use futures::StreamExt; -use log::error; +use log::{debug, 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 @@ -31,6 +31,8 @@ impl Processor { } pub(crate) async fn run(&mut self) { + debug!("starting Dealing Processor"); + while let Some(dealing) = self.receiver.next().await { self.process_dealing(dealing).await } diff --git a/validator-api/src/dkg/events/dispatcher.rs b/validator-api/src/dkg/events/dispatcher.rs index 7269be371c..444597f4ac 100644 --- a/validator-api/src/dkg/events/dispatcher.rs +++ b/validator-api/src/dkg/events/dispatcher.rs @@ -6,8 +6,7 @@ use crate::dkg::events::Event; use crate::dkg::main_loop::ContractEventsSender; use futures::channel::mpsc; use futures::StreamExt; -use log::error; -use log::trace; +use log::{debug, error, trace}; use std::fmt::Display; pub(crate) type DispatcherSender = mpsc::UnboundedSender; @@ -56,6 +55,7 @@ impl Dispatcher { } pub(crate) async fn run(&mut self) { + debug!("starting Dispatcher"); while let Some(new_event) = self.event_receiver.next().await { self.handle_event(new_event) } diff --git a/validator-api/src/dkg/main_loop/mod.rs b/validator-api/src/dkg/main_loop/mod.rs index 8829cab802..d7f683255b 100644 --- a/validator-api/src/dkg/main_loop/mod.rs +++ b/validator-api/src/dkg/main_loop/mod.rs @@ -104,6 +104,11 @@ impl ProcessingLoop { self.dkg_state.try_remove_dealer(dealer_address).await } + async fn process_new_key_submission(&self, height: BlockHeight) { + debug!("attempting to register our own dealer keys for this round of dkg"); + info!(".... but that's not implemented yet...."); + } + async fn process_dealer_changes(&self, changes: Vec, height: BlockHeight) { debug!( "processing dealer set change event with {} changes at height {}", @@ -121,6 +126,7 @@ impl ProcessingLoop { async fn process_event(&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 } @@ -129,6 +135,8 @@ impl ProcessingLoop { } pub(crate) async fn run(&mut self) { + debug!("starting DKG main processing loop"); + while let Some(event) = self.contract_events_receiver.next().await { self.process_event(event).await } diff --git a/validator-api/src/dkg/mod.rs b/validator-api/src/dkg/mod.rs index 62925ad958..f8c0112a10 100644 --- a/validator-api/src/dkg/mod.rs +++ b/validator-api/src/dkg/mod.rs @@ -10,7 +10,7 @@ use crate::Client; use crypto::asymmetric::identity; use dkg::bte; use futures::channel::mpsc; -use log::info; +use log::{error, info}; use rand::rngs::OsRng; use std::net::{SocketAddr, ToSocketAddrs}; use std::sync::Arc; @@ -111,8 +111,7 @@ fn make_client() -> Client { // this one is irrelevant as we don't need to call it let api_url = "http://localhost:8080".parse().unwrap(); - // TODO: this is an incorrect address, but I just wanted to have something compiling - let contract_address = "nymt17x6pt4msccvawgxjeg5nmnygttu56tftg5l6j3" + let contract_address = "nymt14hj2tavq8fpesdwxxcu44rty3hh90vhuysqrsr" .parse() .unwrap(); @@ -160,6 +159,5 @@ pub(crate) async fn dkg_only_main() -> anyhow::Result<()> { identity: ed25519_keys, }; - run_dkg(dkg_config, client).await; - Ok(()) + run_dkg(dkg_config, client).await } diff --git a/validator-api/src/dkg/networking/receiver/listener.rs b/validator-api/src/dkg/networking/receiver/listener.rs index 857f6e712c..941d51b56a 100644 --- a/validator-api/src/dkg/networking/receiver/listener.rs +++ b/validator-api/src/dkg/networking/receiver/listener.rs @@ -5,6 +5,7 @@ use crate::dkg::networking::receiver::handler::ConnectionHandler; use crate::dkg::state::StateAccessor; +use log::debug; use std::fmt::Display; use std::net::SocketAddr; use std::process; @@ -35,6 +36,8 @@ impl Listener { where A: ToSocketAddrs + Display, { + debug!("starting off-chain DKG Listener"); + let listener = match TcpListener::bind(&self.address).await { Ok(listener) => listener, Err(err) => { diff --git a/validator-api/src/dkg/smart_contract/watcher/event.rs b/validator-api/src/dkg/smart_contract/watcher/event.rs index 4d74e9bfcd..91115ff875 100644 --- a/validator-api/src/dkg/smart_contract/watcher/event.rs +++ b/validator-api/src/dkg/smart_contract/watcher/event.rs @@ -34,6 +34,7 @@ pub(crate) enum DealerChange { #[derive(Debug)] pub(crate) enum EventType { + NewKeySubmission, DealerSetChange { changes: Vec }, NewDealingCommitment, } @@ -42,6 +43,7 @@ impl Display for EventType { fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { write!(f, "EventType - ")?; match self { + EventType::NewKeySubmission => write!(f, "NewKeySubmission"), EventType::DealerSetChange { changes } => { write!(f, "DealerSetChange with {} changes", changes.len()) } diff --git a/validator-api/src/dkg/smart_contract/watcher/mod.rs b/validator-api/src/dkg/smart_contract/watcher/mod.rs index b55fe78592..5bc62c0428 100644 --- a/validator-api/src/dkg/smart_contract/watcher/mod.rs +++ b/validator-api/src/dkg/smart_contract/watcher/mod.rs @@ -4,13 +4,14 @@ use crate::dkg::error::DkgError; use crate::dkg::state::StateAccessor; use crate::Client; -use coconut_dkg_common::types::EpochState; +use coconut_dkg_common::types::{BlockHeight, DealerDetails, EpochState}; use log::{debug, trace, warn}; use std::collections::HashMap; use std::time::Duration; use tokio::time::interval; pub(crate) use event::{DealerChange, Event, EventType}; +use mixnet_contract_common::Addr; use validator_client::nymd::SigningCosmWasmClient; mod event; @@ -37,22 +38,17 @@ where } } - async fn check_for_dealers(&self) -> Result<(), DkgError> { + async fn check_for_dealers( + &self, + contract_dealers: HashMap, + current_height: BlockHeight, + ) -> Result<(), DkgError> { // get current state let known_dealers = self.state_accessor.get_known_dealers().await; let bad_dealers = self.state_accessor.get_malformed_dealers().await; - // get data from the contract - let current_height = self.client.current_block_height().await?.value(); - let contract_dealers = self.client.get_current_dealers().await?; - let contract_dealers = contract_dealers - .into_iter() - .map(|dealer| (dealer.address.clone(), dealer)) - .collect::>(); - // TODO: this would probably have to get generalised since we'd need to use the same logic // for other possible events - let mut changes = Vec::new(); // check for removed dealers (if our lists contain keys that do not exist in the contract, @@ -100,9 +96,65 @@ where Ok(()) } + async fn check_for_own_submission( + &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; + } + } + } + + Ok(()) + } + + async fn public_key_submission_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::>(); + + self.check_for_own_submission(&contract_dealers, current_height) + .await?; + self.check_for_dealers(contract_dealers, current_height) + .await + } + async fn perform_epoch_state_based_actions(&self, state: EpochState) -> Result<(), DkgError> { match state { - EpochState::PublicKeySubmission { .. } => self.check_for_dealers().await, + EpochState::PublicKeySubmission { .. } => self.public_key_submission_actions().await, EpochState::DealingExchange { .. } => todo!(), EpochState::ComplaintSubmission { .. } => todo!(), EpochState::ComplaintVoting { .. } => todo!(), @@ -138,6 +190,10 @@ where current_epoch.state, prior_epoch.state ); + // this is not entirely true, but for time being let's just use it to test basic event propagation + self.perform_epoch_state_based_actions(current_epoch.state) + .await?; + if prior_epoch.state != current_epoch.state { todo!() } @@ -146,6 +202,7 @@ where } pub(crate) async fn run(&self) { + debug!("Starting dkg contract poller"); let mut interval = interval(self.polling_rate); loop { interval.tick().await; diff --git a/validator-api/src/dkg/state/accessor.rs b/validator-api/src/dkg/state/accessor.rs index 880490f139..7dab7387e4 100644 --- a/validator-api/src/dkg/state/accessor.rs +++ b/validator-api/src/dkg/state/accessor.rs @@ -4,9 +4,9 @@ use crate::dkg::events::{DispatcherSender, Event}; use crate::dkg::smart_contract::watcher; use crate::dkg::state::{Dealer, DkgState, IdentityBytes, MalformedDealer, ReceivedDealing}; -use coconut_dkg_common::types::{Addr, Epoch}; +use coconut_dkg_common::types::{Addr, BlockHeight, Epoch}; use crypto::asymmetric::identity; -use std::collections::{HashMap, HashSet}; +use std::collections::HashMap; use std::net::SocketAddr; // essentially some intermediary that allows either pushing events to the dispatcher or operating @@ -29,7 +29,7 @@ impl StateAccessor { pub(crate) async fn push_event(&self, event: Event) { if let Err(err) = self.dispatcher_sender.unbounded_send(event) { log::error!("Our event dispatcher failed to receive {} event - it has presumably crashed. Shutting down the API after saving DKG state", err.into_inner()); - self.dkg_state.save().await; + self.dkg_state.save_to_file().await; std::process::exit(1); } } @@ -39,6 +39,18 @@ impl StateAccessor { .await } + pub(crate) async fn push_new_key_submission_event(&self, block_height: BlockHeight) { + self.push_event(Event::new_contract_change_event(watcher::Event::new( + block_height, + watcher::EventType::NewKeySubmission, + ))) + .await + } + + pub(crate) async fn has_submitted_keys(&self) -> bool { + self.dkg_state.has_submitted_keys().await + } + pub(crate) async fn current_epoch(&self) -> Epoch { self.dkg_state.current_epoch().await } diff --git a/validator-api/src/dkg/state/mod.rs b/validator-api/src/dkg/state/mod.rs index db03454a7c..4df5e4fd93 100644 --- a/validator-api/src/dkg/state/mod.rs +++ b/validator-api/src/dkg/state/mod.rs @@ -1,10 +1,12 @@ // Copyright 2022 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 +use crate::Client; use coconut_dkg_common::types::{Addr, BlockHeight, DealerDetails, Epoch, NodeIndex}; use crypto::asymmetric::identity; use dkg::{bte, Dealing}; use futures::lock::Mutex; +use log::debug; use log::error; use serde::{Deserialize, Serialize}; use std::collections::HashMap; @@ -13,7 +15,9 @@ use std::sync::Arc; mod accessor; +use crate::dkg::error::DkgError; pub(crate) use accessor::StateAccessor; +use validator_client::nymd::SigningCosmWasmClient; type IdentityBytes = [u8; identity::PUBLIC_KEY_LENGTH]; @@ -88,11 +92,6 @@ impl MalformedDealer { } } -#[derive(Debug, Clone)] -pub(crate) struct DkgState { - inner: Arc>, -} - #[derive(Debug, Clone, Serialize, Deserialize)] pub struct ReceivedDealing { epoch_id: u32, @@ -100,17 +99,28 @@ pub struct ReceivedDealing { signature: identity::Signature, } +#[derive(Debug, Clone)] +pub(crate) struct DkgState { + inner_state: Arc>, + keys: Arc, +} + +// we don't want to serialize/deserialize those as they are treated differently +#[derive(Debug)] +struct Keys { + identity: identity::KeyPair, + bte_decryption_key: bte::DecryptionKey, + bte_public_key: bte::PublicKeyWithProof, +} + #[derive(Debug, Serialize, Deserialize)] struct DkgStateInner { submitted_keys: bool, submitted_commitment: bool, submitted_verification_keys: bool, assigned_index: NodeIndex, - signing_key: identity::PrivateKey, - signing_public_key: identity::PublicKey, last_seen_height: BlockHeight, - bte_decryption_key: bte::DecryptionKey, current_epoch: Epoch, @@ -126,23 +136,55 @@ struct DkgStateInner { impl DkgState { // this should only ever be called once, during init - pub(crate) fn new_fresh() -> Self { - // DkgState { - // inner: Arc::new(Mutex::new(DkgStateInner { - // // - // })), - // } + pub(crate) async fn initialise_fresh( + nyxd_client: &Client, + identity: identity::KeyPair, + bte_decryption_key: bte::DecryptionKey, + bte_public_key: bte::PublicKeyWithProof, + ) -> Result + where + C: SigningCosmWasmClient + Send + Sync, + { + debug!("attempting to initialise fresh dkg state"); + let current_epoch = nyxd_client.get_dkg_epoch().await?; + + // TODO: IF we didn't load the state from the file, grab all other data from the contract while + // we're at it, like dealers, dealing commitments, etc. + + Ok(DkgState { + inner_state: Arc::new(Mutex::new(DkgStateInner { + submitted_keys: false, + submitted_commitment: false, + submitted_verification_keys: false, + assigned_index: 0, + last_seen_height: 0, + current_epoch, + expected_epoch_dealing_digests: HashMap::new(), + bad_dealers: HashMap::new(), + current_epoch_dealers: HashMap::new(), + verified_epoch_dealings: HashMap::new(), + unconfirmed_dealings: HashMap::new(), + })), + keys: Arc::new(Keys { + identity, + bte_decryption_key, + bte_public_key, + }), + }) + } + + pub(crate) async fn load_from_file(&self) { todo!() } // some save/load action here - pub(crate) async fn save(&self) { + pub(crate) async fn save_to_file(&self) { todo!() } pub(crate) async fn is_dealers_remote_address(&self, remote: SocketAddr) -> (bool, Epoch) { - let guard = self.inner.lock().await; + let guard = self.inner_state.lock().await; let epoch = guard.current_epoch; let dealers = &guard.current_epoch_dealers; @@ -154,15 +196,19 @@ impl DkgState { ) } + pub(crate) async fn has_submitted_keys(&self) -> bool { + self.inner_state.lock().await.submitted_keys + } + pub(crate) async fn current_epoch(&self) -> Epoch { - self.inner.lock().await.current_epoch + self.inner_state.lock().await.current_epoch } pub(crate) async fn get_verified_dealing( &self, dealer: identity::PublicKey, ) -> Option { - self.inner + self.inner_state .lock() .await .verified_epoch_dealings @@ -171,21 +217,21 @@ impl DkgState { } pub(crate) async fn get_known_dealers(&self) -> HashMap { - self.inner.lock().await.current_epoch_dealers.clone() + self.inner_state.lock().await.current_epoch_dealers.clone() } pub(crate) async fn get_malformed_dealers(&self) -> HashMap { - self.inner.lock().await.bad_dealers.clone() + self.inner_state.lock().await.bad_dealers.clone() } pub(crate) async fn update_last_seen_height(&self, new_last_seen: BlockHeight) { - self.inner.lock().await.last_seen_height = new_last_seen; + self.inner_state.lock().await.last_seen_height = new_last_seen; } pub(crate) async fn try_add_new_dealer(&self, dealer: Dealer) { - // TODO: perhaps we should panic or something instead since this should have never occured in the first place? + // TODO: perhaps we should panic or something instead since this should have never occurred in the first place? if let Some(old_dealer) = self - .inner + .inner_state .lock() .await .current_epoch_dealers @@ -199,9 +245,9 @@ impl DkgState { } pub(crate) async fn try_add_malformed_dealer(&self, dealer_details: MalformedDealer) { - // TODO: perhaps we should panic or something instead since this should have never occured in the first place? + // TODO: perhaps we should panic or something instead since this should have never occurred in the first place? if let Some(old_dealer) = self - .inner + .inner_state .lock() .await .bad_dealers @@ -215,7 +261,7 @@ impl DkgState { } pub(crate) async fn try_remove_dealer(&self, dealer_address: Addr) { - let mut guard = self.inner.lock().await; + let mut guard = self.inner_state.lock().await; // dealer is in either bad dealers or known dealers, never both, // so if we managed to remove it from the former, we don't need to check the latter diff --git a/validator-api/src/nymd_client.rs b/validator-api/src/nymd_client.rs index a9964f5ab5..d9ad0f7261 100644 --- a/validator-api/src/nymd_client.rs +++ b/validator-api/src/nymd_client.rs @@ -5,7 +5,9 @@ use crate::config::Config; use crate::rewarded_set_updater::error::RewardingError; #[cfg(feature = "coconut")] use async_trait::async_trait; -use coconut_dkg_common::types::{DealerDetails, Epoch as DkgEpoch}; +use coconut_dkg_common::types::{ + BlacklistedDealer, BlacklistingResponse, DealerDetails, Epoch as DkgEpoch, +}; use config::defaults::{DEFAULT_NETWORK, DEFAULT_VALIDATOR_API_PORT}; use mixnet_contract_common::Interval; use mixnet_contract_common::{ @@ -22,7 +24,7 @@ use validator_client::nymd::error::NymdError; use validator_client::nymd::traits::DkgClient; use validator_client::nymd::{ hash::{Hash, SHA256_HASH_SIZE}, - CosmWasmClient, CosmosCoin, Fee, Height, QueryNymdClient, SigningCosmWasmClient, + AccountId, CosmWasmClient, CosmosCoin, Fee, Height, QueryNymdClient, SigningCosmWasmClient, SigningNymdClient, TendermintTime, }; use validator_client::ValidatorClientError; @@ -280,6 +282,13 @@ impl Client { .await } + pub(crate) async fn address(&self) -> AccountId + where + C: SigningCosmWasmClient + Sync, + { + self.0.read().await.nymd.address().clone() + } + #[allow(dead_code)] pub(crate) async fn advance_current_epoch(&self) -> Result<(), ValidatorClientError> where @@ -457,6 +466,19 @@ where self.0.read().await.get_all_nymd_past_dealers().await } + pub(crate) async fn get_blacklisted_dealers( + &self, + ) -> Result, ValidatorClientError> { + self.0.read().await.get_all_nymd_blacklisted_dealers().await + } + + pub(crate) async fn get_blacklisting( + &self, + address: String, + ) -> Result { + self.0.read().await.nymd.get_blacklisting(address).await + } + pub(crate) async fn submit_dealing_commitment( &self, epoch_id: u32,