diff --git a/primitives/src/tendermint.rs b/primitives/src/tendermint.rs index 7177fbb502..75d861ec87 100644 --- a/primitives/src/tendermint.rs +++ b/primitives/src/tendermint.rs @@ -29,7 +29,7 @@ impl SerializedSize for TendermintStep { } /// Unique identifier for a single instance of TendermintAggregation -#[derive(Serialize, Deserialize, Debug, Clone, Eq, PartialEq)] +#[derive(Copy, Clone, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)] pub struct TendermintIdentifier { /// Network ID this tendermint vote is meant for. pub network: NetworkId, @@ -105,7 +105,7 @@ impl TendermintProposal { // that can be included plain text as the proof alongside it also contains it. #[derive(Debug, Clone, Eq, PartialEq)] pub struct TendermintVote { - /// Hash of the proposed macro block + /// Hash of the proposed macro or skip block pub proposal_hash: Option, /// Identifier to this votes aggregation pub id: TendermintIdentifier, diff --git a/validator/src/aggregation/tendermint/proposal.rs b/validator/src/aggregation/tendermint/proposal.rs index a659758d06..59bf3bbb8e 100644 --- a/validator/src/aggregation/tendermint/proposal.rs +++ b/validator/src/aggregation/tendermint/proposal.rs @@ -4,7 +4,7 @@ use nimiq_block::{MacroBody, MacroHeader, MicroBlock}; use nimiq_blockchain::Blockchain; use nimiq_blockchain_interface::AbstractBlockchain; use nimiq_hash::{Blake2sHash, Hash}; -use nimiq_keys::Ed25519Signature as SchnorrSignature; +use nimiq_keys::{Address, Ed25519Signature as SchnorrSignature}; use nimiq_network_interface::{ network::Network, request::{Handle, RequestCommon, RequestMarker}, @@ -59,10 +59,11 @@ impl SignedProposal { /// via GossipSub, i.e. produced by this node itself. pub fn into_tendermint_signed_message( self, + address: Address, id: Option, - ) -> SignedProposalMessage, (SchnorrSignature, u16)> { + ) -> SignedProposalMessage, (SchnorrSignature, Address, u16)> { SignedProposalMessage { - signature: (self.signature, self.signer), + signature: (self.signature, address, self.signer), message: ProposalMessage { proposal: Header(self.proposal, id), round: self.round, @@ -78,15 +79,15 @@ impl SignedProposal { &self, predecessor: MicroBlock, blockchain: &Blockchain, - ) -> bool { + ) -> Result { // Make sure the proposal references the predecessor as its parent hash if predecessor.hash() != self.proposal.parent_hash { - return false; + return Err(()); } // Make sure the height of the predecessor fits if predecessor.block_number() + 1 != self.proposal.block_number { - return false; + return Err(()); } // Get the active validators. let validators = blockchain.current_validators().unwrap(); @@ -106,18 +107,24 @@ impl SignedProposal { .validator; // Compare the expected and the actual validator - *assumed_validator == actual_validator + if *assumed_validator != actual_validator { + return Err(()); + } + + Ok(actual_validator.address) } } -impl From, (SchnorrSignature, u16)>> for SignedProposal { - fn from(value: SignedProposalMessage, (SchnorrSignature, u16)>) -> Self { +impl From, (SchnorrSignature, Address, u16)>> + for SignedProposal +{ + fn from(value: SignedProposalMessage, (SchnorrSignature, Address, u16)>) -> Self { Self { proposal: value.message.proposal.0, valid_round: value.message.valid_round, round: value.message.round, signature: value.signature.0, - signer: value.signature.1, + signer: value.signature.2, } } } diff --git a/validator/src/aggregation/tendermint/protocol.rs b/validator/src/aggregation/tendermint/protocol.rs index aff073cc5b..e2b0c04dc5 100644 --- a/validator/src/aggregation/tendermint/protocol.rs +++ b/validator/src/aggregation/tendermint/protocol.rs @@ -1,11 +1,12 @@ use std::sync::Arc; +use nimiq_block::MultiSignature; use nimiq_handel::{ evaluator::WeightedVote, partitioner::BinomialPartitioner, protocol::Protocol, store::ReplaceStore, }; -use nimiq_primitives::{policy::Policy, TendermintIdentifier}; -use nimiq_tendermint::Aggregation; +use nimiq_primitives::{policy::Policy, TendermintIdentifier, TendermintVote}; +use nimiq_tendermint::Aggregation as _; use parking_lot::RwLock; use super::{ @@ -13,7 +14,6 @@ use super::{ verifier::TendermintVerifier, }; -#[derive(std::fmt::Debug)] pub(crate) struct TendermintAggregationProtocol { verifier: Arc<>::Verifier>, partitioner: Arc<>::Partitioner>, @@ -30,6 +30,7 @@ impl TendermintAggregationProtocol { registry: Arc, node_id: usize, id: TendermintIdentifier, + observe_valid_vote: Arc, ) -> Self { let partitioner = Arc::new(BinomialPartitioner::new(node_id, registry.len())); @@ -53,7 +54,11 @@ impl TendermintAggregationProtocol { }, )); - let verifier = Arc::new(TendermintVerifier::new(registry.clone(), id.clone())); + let verifier = Arc::new(TendermintVerifier::new( + registry.clone(), + id.clone(), + observe_valid_vote, + )); Self { verifier, diff --git a/validator/src/aggregation/tendermint/state.rs b/validator/src/aggregation/tendermint/state.rs index 94e447ba9b..47b56a9dae 100644 --- a/validator/src/aggregation/tendermint/state.rs +++ b/validator/src/aggregation/tendermint/state.rs @@ -6,7 +6,7 @@ use std::{ use nimiq_block::{MacroBody, MacroHeader}; use nimiq_database_value_derive::DbSerializable; use nimiq_hash::Blake2sHash; -use nimiq_keys::Ed25519Signature as SchnorrSignature; +use nimiq_keys::{Address, Ed25519Signature as SchnorrSignature}; use nimiq_serde::{Deserialize, Serialize}; use nimiq_tendermint::{State as TendermintState, Step}; use nimiq_validator_network::{PubsubId, ValidatorNetwork}; @@ -23,7 +23,8 @@ pub struct MacroState { round_number: u32, step: Step, known_proposals: BTreeMap, - round_proposals: BTreeMap, (SchnorrSignature, u16))>>, + round_proposals: + BTreeMap, (SchnorrSignature, Address, u16))>>, votes: BTreeMap<(u32, Step), Option>, best_votes: BTreeMap<(u32, Step), TendermintContribution>, inherents: BTreeMap, @@ -177,7 +178,7 @@ impl MacroState { round: round_number, valid_round, signature: signature.0, - signer: signature.1, + signer: signature.2, }) } } diff --git a/validator/src/aggregation/tendermint/verifier.rs b/validator/src/aggregation/tendermint/verifier.rs index 1f0cc3c914..c8a97b37bf 100644 --- a/validator/src/aggregation/tendermint/verifier.rs +++ b/validator/src/aggregation/tendermint/verifier.rs @@ -1,6 +1,7 @@ use std::sync::Arc; use async_trait::async_trait; +use nimiq_block::MultiSignature; use nimiq_bls::AggregatePublicKey; use nimiq_handel::{ identity::IdentityRegistry, @@ -13,17 +14,22 @@ use tokio::task; use super::contribution::TendermintContribution; -#[derive(Debug)] pub(crate) struct TendermintVerifier { identity_registry: Arc, id: TendermintIdentifier, + observe_valid_vote: Arc, } impl TendermintVerifier { - pub(crate) fn new(identity_registry: Arc, id: TendermintIdentifier) -> Self { + pub(crate) fn new( + identity_registry: Arc, + id: TendermintIdentifier, + observe_valid_vote: Arc, + ) -> Self { Self { identity_registry, id, + observe_valid_vote, } } } @@ -58,11 +64,13 @@ impl Verifier for TendermintVerifie params.push((aggregated_public_key, vote, multi_sig.clone())); } + let observe_valid_vote = Arc::clone(&self.observe_valid_vote); let result = task::spawn_blocking(move || { params .into_par_iter() .map(|(aggregated_public_key, vote, contribution)| { if aggregated_public_key.verify_hash(vote.hash(), &contribution.signature) { + observe_valid_vote(&vote, &contribution); Ok(()) } else { Err(()) diff --git a/validator/src/double_proposal.rs b/validator/src/double_proposal.rs new file mode 100644 index 0000000000..c7306a5bfc --- /dev/null +++ b/validator/src/double_proposal.rs @@ -0,0 +1,97 @@ +use std::{ + collections::{btree_map, BTreeMap}, + mem, +}; + +use nimiq_block::{DoubleProposalProof, MacroHeader}; +use nimiq_hash::Blake2sHash; +use nimiq_keys::{Address, Ed25519Signature as SchnorrSignature}; +use nimiq_primitives::{networks::NetworkId, TendermintProposal}; + +struct Proposal { + proposer: Address, + hash: Blake2sHash, + proposal: TendermintProposal, + signature: SchnorrSignature, +} + +enum Round { + Seen(Proposal), + Reported(Address), +} + +impl Round { + fn proposer(&self) -> &Address { + match self { + Round::Seen(Proposal { proposer, .. }) => proposer, + Round::Reported(proposer) => proposer, + } + } +} + +pub struct DoubleProposalDetector { + network: NetworkId, + block_number: u32, + rounds: BTreeMap, +} + +impl DoubleProposalDetector { + // TODO: add network_id + // TODO: record one proposal per validator per height + pub fn new( + network: NetworkId, + block_number: u32, + ) -> DoubleProposalDetector { + DoubleProposalDetector { + network, + block_number, + rounds: BTreeMap::new(), + } + } + pub fn observe_valid_proposal( + &mut self, + proposer: Address, + proposal: TendermintProposal, + signature: SchnorrSignature, + ) -> Option { + assert_eq!(proposal.proposal.network, self.network); + assert_eq!(proposal.proposal.block_number, self.block_number); + match self.rounds.entry(proposal.round) { + btree_map::Entry::Vacant(v) => { + v.insert(Round::Seen(Proposal { + proposer, + hash: proposal.hash(), + proposal, + signature, + })); + None + } + btree_map::Entry::Occupied(o) => { + let round = o.into_mut(); + assert_eq!( + *round.proposer(), + proposer, + "Only one address can propose in a round" + ); + match round { + Round::Reported(_) => return None, + Round::Seen(Proposal { hash: old_hash, .. }) => { + if *old_hash == proposal.hash() { + return None; + } + } + } + match mem::replace(round, Round::Reported(proposer.clone())) { + Round::Reported(_) => unreachable!(), + Round::Seen(old_proposal) => Some(DoubleProposalProof::new( + proposer, + old_proposal.proposal, + old_proposal.signature, + proposal, + signature, + )), + } + } + } + } +} diff --git a/validator/src/double_vote.rs b/validator/src/double_vote.rs new file mode 100644 index 0000000000..d5b4f67e3f --- /dev/null +++ b/validator/src/double_vote.rs @@ -0,0 +1,113 @@ +use std::{ + collections::{btree_map, BTreeMap}, + mem, +}; + +use nimiq_block::{DoubleVoteProof, MultiSignature}; +use nimiq_hash::Blake2sHash; +use nimiq_primitives::{ + networks::NetworkId, policy::Policy, slots_allocation::Validators, TendermintIdentifier, + TendermintVote, +}; + +struct Vote { + vote: Option, + signature: MultiSignature, +} + +enum SlotBand { + Seen(Vote), + Reported, +} + +pub struct DoubleVoteDetector { + network: NetworkId, + block_number: u32, + validators: Validators, + round_slot_bands: BTreeMap<(TendermintIdentifier, u16), SlotBand>, +} + +impl DoubleVoteDetector { + pub fn new( + network: NetworkId, + block_number: u32, + validators: Validators, + ) -> DoubleVoteDetector { + DoubleVoteDetector { + network, + block_number, + validators, + round_slot_bands: BTreeMap::new(), + } + } + pub fn observe_valid_vote( + &mut self, + vote: &TendermintVote, + signature: &MultiSignature, + ) -> Vec { + assert_eq!(vote.id.network, self.network); + assert_eq!(vote.id.block_number, self.block_number); + + let &TendermintVote { + id, + proposal_hash: ref vote, + } = vote; + let mut result = Vec::new(); + for slot in signature.signers.iter() { + assert!(slot < Policy::SLOTS as usize); + let slot = u16::try_from(slot).unwrap(); + let slot_band = self.validators.get_band_from_slot(slot); + if let Some(proof) = + self.observe_valid_vote_from(slot_band, id, vote.clone(), signature) + { + result.push(proof) + } + } + result + } + + fn observe_valid_vote_from( + &mut self, + slot_band: u16, + id: TendermintIdentifier, + vote: Option, + signature: &MultiSignature, + ) -> Option { + match self.round_slot_bands.entry((id, slot_band)) { + btree_map::Entry::Vacant(v) => { + v.insert(SlotBand::Seen(Vote { + vote: vote.clone(), + signature: signature.clone(), + })); + None + } + btree_map::Entry::Occupied(o) => { + let entry = o.into_mut(); + match entry { + SlotBand::Reported => return None, + SlotBand::Seen(Vote { vote: old_vote, .. }) => { + if *old_vote == vote { + return None; + } + } + } + match mem::replace(entry, SlotBand::Reported) { + SlotBand::Reported => unreachable!(), + SlotBand::Seen(old) => Some(DoubleVoteProof::new( + id, + self.validators + .get_validator_by_slot_band(slot_band) + .address + .clone(), + old.vote, + old.signature.signature, + old.signature.signers, + vote.clone(), + signature.signature, + signature.signers.clone(), + )), + } + } + } + } +} diff --git a/validator/src/lib.rs b/validator/src/lib.rs index 19fe7e0c81..2bd0a4cf21 100644 --- a/validator/src/lib.rs +++ b/validator/src/lib.rs @@ -2,6 +2,8 @@ extern crate log; pub mod aggregation; +mod double_proposal; +mod double_vote; mod jail; pub mod key_utils; mod r#macro; diff --git a/validator/src/macro.rs b/validator/src/macro.rs index a76f81d066..c99eb70e8b 100644 --- a/validator/src/macro.rs +++ b/validator/src/macro.rs @@ -6,16 +6,19 @@ use std::{ }; use futures::stream::{BoxStream, Stream, StreamExt}; -use nimiq_block::MacroBlock; +use nimiq_block::{DoubleProposalProof, DoubleVoteProof, MacroBlock, MacroHeader, MultiSignature}; use nimiq_blockchain::{BlockProducer, Blockchain}; -use nimiq_keys::Ed25519Signature as SchnorrSignature; +use nimiq_blockchain_interface::AbstractBlockchain; +use nimiq_keys::{Address, Ed25519Signature as SchnorrSignature}; use nimiq_network_interface::network::Topic; -use nimiq_primitives::{networks::NetworkId, slots_allocation::Validators}; +use nimiq_primitives::{ + networks::NetworkId, slots_allocation::Validators, TendermintProposal, TendermintVote, +}; use nimiq_tendermint::{ Return as TendermintReturn, SignedProposalMessage, TaggedAggregationMessage, Tendermint, }; use nimiq_validator_network::{PubsubId, ValidatorNetwork}; -use parking_lot::RwLock; +use parking_lot::{Mutex, RwLock}; use crate::{ aggregation::tendermint::{ @@ -24,6 +27,8 @@ use crate::{ state::MacroState, update_message::TendermintUpdate, }, + double_proposal::DoubleProposalDetector, + double_vote::DoubleVoteDetector, tendermint::TendermintProtocol, }; @@ -34,13 +39,22 @@ where Update(MacroState), Decision(MacroBlock), ProposalAccepted( - SignedProposalMessage>, (SchnorrSignature, u16)>, + SignedProposalMessage< + Header>, + (SchnorrSignature, Address, u16), + >, ), ProposalIgnored( - SignedProposalMessage>, (SchnorrSignature, u16)>, + SignedProposalMessage< + Header>, + (SchnorrSignature, Address, u16), + >, ), ProposalRejected( - SignedProposalMessage>, (SchnorrSignature, u16)>, + SignedProposalMessage< + Header>, + (SchnorrSignature, Address, u16), + >, ), } @@ -83,10 +97,16 @@ where state_opt: Option, proposal_stream: BoxStream< 'static, - SignedProposalMessage>, (SchnorrSignature, u16)>, + SignedProposalMessage< + Header>, + (SchnorrSignature, Address, u16), + >, >, + // TODO: make this just one parameter for equivocation proofs + on_double_proposal: Arc, + on_double_vote: Arc, ) -> Self { - let input = network + let level_update_stream = network .receive::() .filter_map(move |(item, validator_id)| async move { // Check that the update is for the correct block. @@ -100,6 +120,55 @@ where }) .boxed(); + let observe_valid_vote = { + let double_vote_detector = Arc::new(Mutex::new(DoubleVoteDetector::new( + network_id, + block_height, + blockchain.read().current_validators().unwrap().clone(), + ))); + Arc::new(move |vote: &TendermintVote, signature: &MultiSignature| { + for proof in double_vote_detector + .lock() + .observe_valid_vote(vote, signature) + { + on_double_vote(proof); + } + }) + }; + + let observe_valid_requested_proposal = { + let double_proposal_detector = + Arc::new(Mutex::new(DoubleProposalDetector::new(network_id, block_height))); + Arc::new( + move |address: Address, + proposal: TendermintProposal, + signature: SchnorrSignature| { + if let Some(proof) = double_proposal_detector + .lock() + // TODO: are these already verified here? + .observe_valid_proposal(address, proposal, signature) + { + on_double_proposal(proof); + } + }, + ) + }; + + let observe = Arc::clone(&observe_valid_requested_proposal); + let proposal_stream = proposal_stream + .inspect(move |proposal| { + observe( + proposal.signature.1.clone(), + TendermintProposal { + proposal: proposal.message.proposal.0.clone(), + round: proposal.message.round, + valid_round: proposal.message.valid_round, + }, + proposal.signature.0.clone(), + ) + }) + .boxed(); + let dependencies = TendermintProtocol::new( blockchain, network, @@ -108,6 +177,8 @@ where validator_slot_band, network_id, block_height, + observe_valid_requested_proposal, + observe_valid_vote, ); // create the Tendermint instance, which implements Stream @@ -115,7 +186,7 @@ where dependencies, state_opt.and_then(|s| s.into_tendermint_state(block_height)), proposal_stream, - input, + level_update_stream, ) // and map the return value such that a state update can be persisted. .map(move |item| match item { diff --git a/validator/src/proposal_buffer.rs b/validator/src/proposal_buffer.rs index a5117e4bf0..91e68ece75 100644 --- a/validator/src/proposal_buffer.rs +++ b/validator/src/proposal_buffer.rs @@ -16,7 +16,7 @@ use nimiq_blockchain_interface::AbstractBlockchain; use nimiq_consensus::consensus::{ consensus_proxy::ConsensusProxy, ResolveBlockError as ConsensusResolveBlockError, }; -use nimiq_keys::Ed25519Signature as SchnorrSignature; +use nimiq_keys::{Address, Ed25519Signature as SchnorrSignature}; use nimiq_network_interface::network::{CloseReason, MsgAcceptance, Network, PubsubId as _, Topic}; use nimiq_primitives::{policy::Policy, TendermintProposal}; use nimiq_serde::Serialize; @@ -62,10 +62,7 @@ enum ResolveBlockError { /// the assumed proposer may be punished (as he produced faulty data, proven by the signature and the predecessor vrf). /// If the predecessor is unavailable (even after requesting it from the peer who originated the proposal or relayed /// the proposal) they both could be banned as they failed to produce proper data upon being asked to do so. -pub(crate) struct ProposalBuffer -where - PubsubId: std::fmt::Debug + Unpin, -{ +pub(crate) struct ProposalBuffer { /// The network used to validate messages and disconnect peers if necessary. network: Arc, @@ -88,7 +85,7 @@ where BoxFuture< 'static, Result< - (SignedProposal, PubsubId), + (SignedProposal, Address, PubsubId), ResolveBlockError, >, >, @@ -103,10 +100,7 @@ where waker: Option, } -impl ProposalBuffer -where - PubsubId: std::fmt::Debug + Unpin, -{ +impl ProposalBuffer { /// Creates a new ProposalBuffer, returning the [ProposalSender] and [ProposalReceiver] that share the buffer. /// Blockchain, Consensus and Network are necessary to do basic verification and punishments as well as to resolve blocks. // Ignoring clippy warning: this return type is on purpose @@ -182,13 +176,13 @@ where pub fn poll_resolve_block_futures( &mut self, cx: &mut Context, - ) -> Option> { + ) -> Option<(SignedProposal, Address, PubsubId)> { while let Poll::Ready(Some(result)) = self.resolve_block_futures.poll_next_unpin(cx) { match result { Ok(proposal_and_id) => { // Proposal is good to go. Remove peer from the map and return. self.peers_with_resolving_blocks - .remove(&proposal_and_id.1.propagation_source()); + .remove(&proposal_and_id.2.propagation_source()); return Some(proposal_and_id); } Err(ResolveBlockError::Invalid(pubsub_id)) => { @@ -227,7 +221,7 @@ where pub fn poll_proposal( &mut self, blockchain_arc: &Arc>, - ) -> Option<(SignedProposal, PubsubId)> { + ) -> Option<(SignedProposal, Address, PubsubId)> { while let Some((_peer, (signed_proposal, pubsub_id))) = self.buffer.pop_front() { // Get a read lock of the blockchain. let blockchain = blockchain_arc.read(); @@ -251,15 +245,17 @@ where } // Micro block predecessors can be used to verify the signer. If the block itself is good will be checked later. Ok(Block::Micro(block)) => { - if !signed_proposal.verify_signer_matches_producer(block, &blockchain) { + if let Ok(address) = + signed_proposal.verify_signer_matches_producer(block, &blockchain) + { + // No validate message call here, as later in the process more proposal verification happens. + return Some((signed_proposal, address, pubsub_id)); + } else { log::debug!( ?pubsub_id, "Verification of signed proposal failed. Disconnecting the peer." ); self.disconnect_and_reject(pubsub_id); - } else { - // No validate message call here, as later in the process more proposal verification happens. - return Some((signed_proposal, pubsub_id)); } } Err(_error) => { @@ -293,8 +289,10 @@ where (signed_proposal, pubsub_id): (SignedProposal, PubsubId), consensus_proxy: ConsensusProxy, blockchain: Arc>, - ) -> Result<(SignedProposal, PubsubId), ResolveBlockError> - { + ) -> Result< + (SignedProposal, Address, PubsubId), + ResolveBlockError, + > { let hash = signed_proposal.proposal.parent_hash.clone(); consensus_proxy @@ -320,9 +318,10 @@ where let blockchain = blockchain.read(); // Make sure the signer matches the producer. This also performs some basic predecessor checks. - if signed_proposal.verify_signer_matches_producer(predecessor, &blockchain) + if let Ok(address) = + signed_proposal.verify_signer_matches_producer(predecessor, &blockchain) { - Ok((signed_proposal, pubsub_id)) + Ok((signed_proposal, address, pubsub_id)) } else { Err(ResolveBlockError::Invalid(pubsub_id)) } @@ -337,10 +336,7 @@ where /// identity in the message. /// Checking for a known predecessor and it having been signed by the correct proposer, happens on the receiver /// side as chances are higher to already have received the blocks predecessor later in the process. -pub(crate) struct ProposalSender -where - PubsubId: std::fmt::Debug + Unpin, -{ +pub(crate) struct ProposalSender { /// The buffer holding all buffered proposals shared with the [ProposalReceiver] shared: Arc>>, @@ -354,10 +350,7 @@ where network: Arc, } -impl ProposalSender -where - PubsubId: std::fmt::Debug + Unpin, -{ +impl ProposalSender { /// Sends the proposal and PubsubId into the buffer. /// /// This function may lead to the proposal not actually being admitted into the buffer as the signature may not verify. @@ -488,10 +481,7 @@ where } } -pub(crate) struct ProposalReceiver -where - PubsubId: std::fmt::Debug + Unpin, -{ +pub(crate) struct ProposalReceiver { /// The buffer holding all buffered proposals shared with the [ProposalSender] shared: Arc>>, @@ -504,27 +494,27 @@ where network: Arc, } -impl Stream for ProposalReceiver -where - PubsubId: std::fmt::Debug + Unpin, -{ - type Item = SignedProposalMessage>, (SchnorrSignature, u16)>; +impl Stream for ProposalReceiver { + type Item = SignedProposalMessage< + Header>, + (SchnorrSignature, Address, u16), + >; fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { // Acquire the shared buffer lock. let mut shared = self.shared.lock(); // Poll proposals from the shared buffer. - if let Some((proposal, pubsub_id)) = shared.poll_proposal(&self.blockchain) { + if let Some((proposal, address, pubsub_id)) = shared.poll_proposal(&self.blockchain) { return Poll::Ready(Some( - proposal.into_tendermint_signed_message(Some(pubsub_id)), + proposal.into_tendermint_signed_message(address, Some(pubsub_id)), )); } // Poll the resolve block futures to see if a proposals predecessor has resolved. - if let Some((proposal, pubsub_id)) = shared.poll_resolve_block_futures(cx) { + if let Some((proposal, address, pubsub_id)) = shared.poll_resolve_block_futures(cx) { return Poll::Ready(Some( - proposal.into_tendermint_signed_message(Some(pubsub_id)), + proposal.into_tendermint_signed_message(address, Some(pubsub_id)), )); } @@ -544,10 +534,7 @@ where } } -impl Clone for ProposalReceiver -where - PubsubId: std::fmt::Debug + Unpin, -{ +impl Clone for ProposalReceiver { fn clone(&self) -> Self { Self { shared: Arc::clone(&self.shared), @@ -568,7 +555,7 @@ mod test { use nimiq_consensus::{ sync::syncer_proxy::SyncerProxy, BlsCache, Consensus, ConsensusEvent, ConsensusProxy, }; - use nimiq_keys::{KeyPair as SchnorrKeyPair, PrivateKey as SchnorrPrivateKey}; + use nimiq_keys::{Address, KeyPair as SchnorrKeyPair, PrivateKey as SchnorrPrivateKey}; use nimiq_network_interface::network::Network as NetworkInterface; use nimiq_network_mock::{MockHub, MockNetwork}; use nimiq_primitives::{policy::Policy, TendermintProposal}; @@ -710,7 +697,7 @@ mod test { let signed_message = SignedProposalMessage { message: proposal_message, - signature: (signing_key.sign(&data), 0), + signature: (signing_key.sign(&data), Address::default(), 0), }; // Send the proposal over gossipsub to get it correctly filled with a pubsub_id diff --git a/validator/src/tendermint.rs b/validator/src/tendermint.rs index eadd9d3838..388c357eb8 100644 --- a/validator/src/tendermint.rs +++ b/validator/src/tendermint.rs @@ -4,7 +4,7 @@ use futures::{ future::{self, BoxFuture, FutureExt}, stream::{BoxStream, StreamExt}, }; -use nimiq_block::{Block, MacroBlock, TendermintProof}; +use nimiq_block::{Block, MacroBlock, MacroHeader, MultiSignature, TendermintProof}; use nimiq_blockchain::{BlockProducer, Blockchain}; use nimiq_blockchain_interface::AbstractBlockchain; use nimiq_collections::BitSet; @@ -16,7 +16,7 @@ use nimiq_handel::{ verifier::{VerificationResult, Verifier}, }; use nimiq_hash::{Blake2sHash, Hash}; -use nimiq_keys::Ed25519Signature as SchnorrSignature; +use nimiq_keys::{Address, Ed25519Signature as SchnorrSignature}; use nimiq_network_interface::network::CloseReason; use nimiq_primitives::{ networks::NetworkId, policy::Policy, slots_allocation::Validators, TendermintIdentifier, @@ -129,6 +129,9 @@ pub struct TendermintProtocol { blockchain: Arc>, // Validator registry on the heap for easy cloning into handel protocol. validator_registry: Arc, + observe_valid_requested_proposal: + Arc, SchnorrSignature) + Send + Sync>, + observe_valid_vote: Arc, } impl Clone for TendermintProtocol { @@ -142,6 +145,8 @@ impl Clone for TendermintProtocol, SchnorrSignature) + Send + Sync, + >, + observe_valid_vote: Arc, ) -> Self { Self { block_producer, @@ -168,6 +177,8 @@ where validator_registry: Arc::new(ValidatorRegistry::new(current_validators.clone())), current_validators, network, + observe_valid_requested_proposal, + observe_valid_vote, } } } @@ -184,7 +195,7 @@ where type InherentHash = Blake2sHash; type Aggregation = TendermintContribution; type AggregationMessage = AggregateMessage; - type ProposalSignature = (SchnorrSignature, u16); + type ProposalSignature = (SchnorrSignature, Address, u16); const F_PLUS_ONE: usize = Policy::F_PLUS_ONE as usize; const TWO_F_PLUS_ONE: usize = Policy::TWO_F_PLUS_ONE as usize; @@ -285,6 +296,7 @@ where SingleResponseRequester::new(Arc::clone(&self.network), candidate_peers, request, 3, { let blockchain = Arc::clone(&self.blockchain); let block_height = self.block_height; + let observe = Arc::clone(&self.observe_valid_requested_proposal); Box::new(move |response| { if let Some(signed_proposal) = response { let blockchain = blockchain.read(); @@ -308,8 +320,7 @@ where None, ) .expect("Couldn't find slot owner!") - .validator - .signing_key; + .validator; let data = TendermintProposal { proposal: &signed_proposal.proposal, @@ -319,8 +330,22 @@ where .hash() .serialize_to_vec(); - if proposer.verify(&signed_proposal.signature, &data) { - return Some(signed_proposal.into_tendermint_signed_message(None)); + if proposer + .signing_key + .verify(&signed_proposal.signature, &data) + { + observe( + proposer.address.clone(), + TendermintProposal { + proposal: signed_proposal.proposal.clone(), + round: signed_proposal.round, + valid_round: signed_proposal.valid_round, + }, + signed_proposal.signature.clone(), + ); + return Some( + signed_proposal.into_tendermint_signed_message(proposer.address, None), + ); } } None @@ -372,6 +397,9 @@ where .serialize_to_vec(); ( self.block_producer.signing_key.sign(&data), + self.current_validators.validators[usize::from(self.validator_slot_band)] + .address + .clone(), self.validator_slot_band, ) } @@ -427,6 +455,7 @@ where Arc::clone(&self.validator_registry), self.validator_slot_band as usize, id, + Arc::clone(&self.observe_valid_vote), ); Aggregation::new( @@ -462,6 +491,7 @@ where Arc::clone(&self.validator_registry), self.validator_slot_band as usize, id, + Arc::clone(&self.observe_valid_vote), ); async move { diff --git a/validator/src/validator.rs b/validator/src/validator.rs index d145f5131b..20fdaf035a 100644 --- a/validator/src/validator.rs +++ b/validator/src/validator.rs @@ -430,6 +430,28 @@ where next_block_number, self.macro_state.read().clone(), proposal_stream, + { + let blockchain = Arc::clone(&self.blockchain); + let state = Arc::clone(&self.state); + Arc::new(move |double_proposal_proof| { + Self::on_equivocation_proof_impl( + &blockchain, + &state, + double_proposal_proof.into(), + ) + }) + }, + { + let blockchain = Arc::clone(&self.blockchain); + let state = Arc::clone(&self.state); + Arc::new(move |double_vote_proof| { + Self::on_equivocation_proof_impl( + &blockchain, + &state, + double_vote_proof.into(), + ) + }) + }, )); } BlockType::Micro => { @@ -526,22 +548,30 @@ where } } - fn on_equivocation_proof(&mut self, proof: EquivocationProof) { + fn on_equivocation_proof_impl( + blockchain: &RwLock, + validator_state: &RwLock, + proof: EquivocationProof, + ) { // Keep the lock until the proof is added to the proof pool. - let blockchain = self.blockchain.read(); + let blockchain = blockchain.read(); if blockchain .history_store .has_equivocation_proof(proof.locator(), None) { return; } - self.state + validator_state .write() .consensus .equivocation_proofs .insert(proof); } + fn on_equivocation_proof(&mut self, proof: EquivocationProof) { + Self::on_equivocation_proof_impl(&self.blockchain, &self.state, proof); + } + fn poll_macro(&mut self, cx: &mut Context<'_>) { while let Poll::Ready(Some(event)) = self.macro_producer.as_mut().unwrap().poll_next_unpin(cx) diff --git a/validator/tests/tendermint.rs b/validator/tests/tendermint.rs index d35925cf5e..507786785f 100644 --- a/validator/tests/tendermint.rs +++ b/validator/tests/tendermint.rs @@ -119,6 +119,8 @@ async fn it_verifies_inferior_chain_proposals() { 0, NetworkId::UnitAlbatross, blockchain2.read().head().block_number() + 1, + Arc::new(|_, _, _| {}), + Arc::new(|_, _| {}), ); // Make sure the main chain proposal is acceptable.