diff --git a/ethexe/.ethexe.example.local.toml b/ethexe/.ethexe.example.local.toml index b804d792a79..248f9b438b8 100644 --- a/ethexe/.ethexe.example.local.toml +++ b/ethexe/.ethexe.example.local.toml @@ -268,15 +268,19 @@ block-time = 1 # persistent-peers = [] # Path to a JSON file mapping validator Ethereum addresses to their -# Malachite secp256k1 public keys. The Router contract stores the -# validator set as Ethereum addresses; Malachite needs the matching -# public keys to verify votes and proposals. -# File format (both fields are 0x-prefixed hex): +# Malachite secp256k1 public keys and libp2p peer IDs. The Router +# contract stores the validator set as Ethereum addresses; Malachite +# needs the matching public keys to verify votes/proposals and peer IDs +# to reject proposal parts from non-validator publishers before buffering. +# File format (addresses and keys are 0x-prefixed hex, peer IDs are +# libp2p peer-id strings): # ```json # { -# "0xaaaa...": "0x02bbbb...", -# "0xcccc...": "0x03dddd..." +# "0xaaaa...": { +# "public_key": "0x02bbbb...", +# "peer_id": "16Uiu2HAm..." +# } # } # ``` # (optional, default: None — required on validator nodes). -# validator-pub-keys = "/path/to/validators.json" +# validator-identities = "/path/to/validators.json" diff --git a/ethexe/.ethexe.example.toml b/ethexe/.ethexe.example.toml index 09879421fdb..e0d81c52b8f 100644 --- a/ethexe/.ethexe.example.toml +++ b/ethexe/.ethexe.example.toml @@ -268,15 +268,19 @@ # persistent-peers = [] # Path to a JSON file mapping validator Ethereum addresses to their -# Malachite secp256k1 public keys. The Router contract stores the -# validator set as Ethereum addresses; Malachite needs the matching -# public keys to verify votes and proposals. -# File format (both fields are 0x-prefixed hex): +# Malachite secp256k1 public keys and libp2p peer IDs. The Router +# contract stores the validator set as Ethereum addresses; Malachite +# needs the matching public keys to verify votes/proposals and peer IDs +# to reject proposal parts from non-validator publishers before buffering. +# File format (addresses and keys are 0x-prefixed hex, peer IDs are +# libp2p peer-id strings): # ```json # { -# "0xaaaa...": "0x02bbbb...", -# "0xcccc...": "0x03dddd..." +# "0xaaaa...": { +# "public_key": "0x02bbbb...", +# "peer_id": "16Uiu2HAm..." +# } # } # ``` # (optional, default: None — required on validator nodes). -# validator-pub-keys = "/path/to/validators.json" +# validator-identities = "/path/to/validators.json" diff --git a/ethexe/cli/src/params/malachite.rs b/ethexe/cli/src/params/malachite.rs index e33d61d2625..a6305104247 100644 --- a/ethexe/cli/src/params/malachite.rs +++ b/ethexe/cli/src/params/malachite.rs @@ -10,11 +10,11 @@ use super::MergeParams; use anyhow::{Context, Result}; use clap::Parser; -use ethexe_malachite::{MalachiteConfig, Multiaddr}; -use ethexe_service::config::MalachiteCliConfig; +use ethexe_malachite::{MalachiteConfig, Multiaddr, PeerId}; +use ethexe_service::config::{MalachiteCliConfig, ValidatorIdentity}; use gsigner::secp256k1::{Address, PublicKey}; use serde::Deserialize; -use std::{collections::BTreeMap, net::SocketAddr, path::PathBuf}; +use std::{collections::BTreeMap, net::SocketAddr, path::PathBuf, str::FromStr}; /// Parameters for the Malachite consensus service. /// @@ -48,25 +48,33 @@ pub struct MalachiteParams { pub malachite_persistent_peers: Vec, /// Path to a JSON file mapping validator Ethereum addresses to - /// their Malachite secp256k1 public keys. + /// their Malachite secp256k1 public keys and libp2p peer IDs. /// /// The Router contract stores the validator set as Ethereum /// addresses; the Malachite engine needs the matching public - /// keys to verify votes and proposals. At startup, the service - /// loads this table and looks every on-chain validator address - /// up in it (in router order) to build the final validator set. + /// keys to verify votes/proposals and the peer IDs to drop + /// proposal parts from non-validator publishers before buffering. + /// At startup, the service loads this table and looks every + /// on-chain validator address up in it (in router order) to + /// build the final validator set. /// - /// File format (a flat JSON object — both address and key are - /// hex-encoded with `0x` prefix): + /// File format (a flat JSON object — addresses and keys are + /// hex-encoded with `0x` prefix, peer IDs are libp2p peer-id + /// strings): /// ```json /// { - /// "0xaaaa...": "0x02bbbb...", - /// "0xcccc...": "0x03dddd..." + /// "0xaaaa...": { + /// "public_key": "0x02bbbb...", + /// "peer_id": "16Uiu2HAm..." + /// } /// } /// ``` - #[arg(long = "validators-malachite-pub-keys", aliases = &["mala-validator-keys"])] - #[serde(rename = "validator-pub-keys")] - pub validators_malachite_pub_keys: Option, + #[arg( + long = "validators-malachite-identities", + aliases = &["mala-validator-identities"] + )] + #[serde(rename = "validator-identities")] + pub validators_malachite_identities: Option, } impl MalachiteParams { @@ -74,8 +82,8 @@ impl MalachiteParams { /// [`MalachiteCliConfig`]. Missing fields fall back to sensible /// defaults from [`MalachiteConfig`]. pub fn into_config(self) -> Result { - let validator_pub_keys = match self.validators_malachite_pub_keys { - Some(path) => load_validator_pub_keys_table(&path)?, + let validator_identities = match self.validators_malachite_identities { + Some(path) => load_validator_identities_table(&path)?, None => BTreeMap::new(), }; Ok(MalachiteCliConfig { @@ -83,27 +91,60 @@ impl MalachiteParams { .malachite_listen_addr .unwrap_or(MalachiteConfig::DEFAULT_LISTEN_ADDR), persistent_peers: self.malachite_persistent_peers, - validator_pub_keys, + validator_identities, }) } } -/// Read a JSON file with the validator-pubkey table. The map is -/// `{ "0x
": "0x" }`. Errors include the file path -/// for easier diagnosis. -fn load_validator_pub_keys_table(path: &std::path::Path) -> Result> { +#[derive(Deserialize)] +struct RawValidatorIdentity { + public_key: PublicKey, + peer_id: String, +} + +/// Read a JSON file with the validator identity table. The map is +/// `{ "0x
": { "public_key": "0x", "peer_id": "" } }`. +/// Errors include the file path and offending validator for easier diagnosis. +fn load_validator_identities_table( + path: &std::path::Path, +) -> Result> { let content = std::fs::read_to_string(path).with_context(|| { format!( - "failed to read malachite validator pub keys file at {}", + "failed to read malachite validator identities file at {}", path.display() ) })?; - serde_json::from_str(&content).with_context(|| { - format!( - "failed to parse malachite validator pub keys file at {}", - path.display() - ) - }) + let raw: BTreeMap = serde_json::from_str(&content) + .with_context(|| { + format!( + "failed to parse malachite validator identities file at {}", + path.display() + ) + })?; + let mut identities = BTreeMap::new(); + let mut peer_ids = BTreeMap::new(); + for (addr, raw) in raw { + let peer_id = PeerId::from_str(&raw.peer_id).with_context(|| { + format!( + "validator address {addr} has malformed peer_id in {}", + path.display() + ) + })?; + if let Some(previous) = peer_ids.insert(peer_id, addr) { + anyhow::bail!( + "duplicate malachite peer_id {peer_id} in {} for validators {previous} and {addr}", + path.display() + ); + } + identities.insert( + addr, + ValidatorIdentity { + public_key: raw.public_key, + peer_id, + }, + ); + } + Ok(identities) } impl MergeParams for MalachiteParams { @@ -115,9 +156,87 @@ impl MergeParams for MalachiteParams { Self { malachite_listen_addr: self.malachite_listen_addr.or(with.malachite_listen_addr), malachite_persistent_peers: persistent_peers, - validators_malachite_pub_keys: self - .validators_malachite_pub_keys - .or(with.validators_malachite_pub_keys), + validators_malachite_identities: self + .validators_malachite_identities + .or(with.validators_malachite_identities), } } } + +#[cfg(test)] +mod tests { + use super::*; + use ethexe_malachite::malachite_libp2p_peer_id; + use gsigner::schemes::secp256k1::PrivateKey; + use std::io::Write; + use tempfile::NamedTempFile; + + fn test_identity(seed: u8) -> (Address, PublicKey, PeerId) { + let mut bytes = [0u8; 32]; + bytes[31] = seed; + let private_key = PrivateKey::from_seed(bytes).expect("private key"); + ( + private_key.public_key().to_address(), + private_key.public_key(), + malachite_libp2p_peer_id(&private_key.to_bytes()), + ) + } + + fn identity_json(entries: &[(Address, PublicKey, PeerId)]) -> String { + let entries = entries + .iter() + .map(|(addr, public_key, peer_id)| { + format!(r#""{addr}":{{"public_key":"{public_key}","peer_id":"{peer_id}"}}"#) + }) + .collect::>() + .join(","); + format!("{{{entries}}}") + } + + fn temp_identity_file(contents: &str) -> NamedTempFile { + let mut file = NamedTempFile::new().expect("temp identity file"); + file.write_all(contents.as_bytes()) + .expect("write identity file"); + file + } + + #[test] + fn validator_identities_parse_valid_input() { + let one = test_identity(1); + let two = test_identity(2); + let file = temp_identity_file(&identity_json(&[one, two])); + + let identities = load_validator_identities_table(file.path()).expect("valid identities"); + + assert_eq!(identities.len(), 2); + assert_eq!(identities[&one.0].public_key, one.1); + assert_eq!(identities[&one.0].peer_id, one.2); + assert_eq!(identities[&two.0].public_key, two.1); + assert_eq!(identities[&two.0].peer_id, two.2); + } + + #[test] + fn validator_identities_reject_malformed_peer_id() { + let (addr, public_key, _) = test_identity(1); + let file = temp_identity_file(&format!( + r#"{{"{addr}":{{"public_key":"{public_key}","peer_id":"not-a-peer-id"}}}}"# + )); + + let error = load_validator_identities_table(file.path()).unwrap_err(); + + assert!(error.to_string().contains("malformed peer_id")); + assert!(error.to_string().contains(&addr.to_string())); + } + + #[test] + fn validator_identities_reject_duplicate_peer_id() { + let one = test_identity(1); + let mut two = test_identity(2); + two.2 = one.2; + let file = temp_identity_file(&identity_json(&[one, two])); + + let error = load_validator_identities_table(file.path()).unwrap_err(); + + assert!(error.to_string().contains("duplicate malachite peer_id")); + } +} diff --git a/ethexe/malachite/core/src/app.rs b/ethexe/malachite/core/src/app.rs index f9ca209a56b..db92d3bf492 100644 --- a/ethexe/malachite/core/src/app.rs +++ b/ethexe/malachite/core/src/app.rs @@ -55,7 +55,7 @@ use malachitebft_app_channel::{ use malachitebft_core_types::Height as _; use parity_scale_codec::{Decode, Encode}; use std::{ops::RangeInclusive, sync::Arc}; -use tracing::{error, info, warn}; +use tracing::{debug, error, info, warn}; /// Max allowed distance into the future for pending proposal parts. const FUTURE_HEIGHT_WINDOW: u64 = 4; @@ -186,6 +186,13 @@ where // ReceivedProposalPart (we are not proposer) AppMsg::ReceivedProposalPart { from, part, reply } => { + if !self.state.is_active_validator_peer(&from) { + debug!(%from, "Dropping ReceivedProposalPart from non-validator peer"); + if reply.send(None).is_err() { + error!("ReceivedProposalPart: failed to send reply"); + } + return Ok(()); + } let part_type = match &part.content { StreamContent::Data(p) => p.get_type(), StreamContent::Fin => "fin", @@ -420,13 +427,15 @@ where // TODO: #5475 add per-peer token-bucket rate limit before `ingest_proposal_part` // (CPU/bandwidth bound; complements the memory bound from #5473). - // TODO: #5480 gate `from` against a validator-peer-id allowlist so random - // gossip-mesh peers can't reach this code path at all. async fn process_received_proposal_part( &mut self, from: PeerId, part: StreamMessage, ) -> Result>> { + if !self.state.is_active_validator_peer(&from) { + debug!(%from, "Dropping ReceivedProposalPart from non-validator peer"); + return Ok(None); + } let Some(parts) = self.state.ingest_proposal_part(from, part) else { return Ok(None); }; @@ -769,7 +778,7 @@ mod tests { use crate::{ context::{ProposalData, ProposalInit, Validator, ValidatorSet, Value}, signing::{MalachiteSigner, libp2p_peer_id, private_key_from_bytes}, - state::SharedValidatorSet, + state::{SharedValidatorPeers, SharedValidatorSet}, store::Store, }; use async_trait::async_trait; @@ -777,7 +786,7 @@ mod tests { ConsensusRequest, NetworkRequest, app::{events::TxEvent, streaming::StreamId}, }; - use std::time::Duration; + use std::{collections::BTreeSet, time::Duration}; use tempfile::TempDir; use tokio::sync::mpsc; @@ -815,6 +824,10 @@ mod tests { PeerId::from_bytes(&lp.to_bytes()).expect("valid multihash") } + fn validator_peers(peers: impl IntoIterator) -> SharedValidatorPeers { + SharedValidatorPeers::new(peers.into_iter().collect::>()) + } + /// Init + Data + Fin for a fully-formed stream at `height`. The /// proposer field is filled in so [`ProposalParts`] assembles /// without complaint. @@ -862,6 +875,7 @@ mod tests { let mut state = State::::new( signer, validator_set, + validator_peers([test_peer(2)]), address, store, Duration::from_secs(1), @@ -963,6 +977,111 @@ mod tests { ); } + #[tokio::test] + async fn non_validator_proposal_part_is_dropped_before_buffering() { + let current = 10u64; + let (mut handler, _dir, addr) = make_handler(current); + let rogue_peer = test_peer(3); + let near = current + FUTURE_HEIGHT_WINDOW; + + let value = run_stream( + &mut handler, + rogue_peer, + complete_stream(addr, near, b"ignored"), + ) + .await; + + assert!(value.is_none()); + assert_eq!( + handler.state.open_proposal_streams(), + 0, + "non-validator peer must not open a proposal stream" + ); + let pending = handler + .state + .store + .get_pending_proposal_parts(Height::new(near), Round::new(0)) + .unwrap(); + assert!( + pending.is_empty(), + "non-validator peer must not create pending proposal DB writes" + ); + } + + #[tokio::test] + async fn validator_peer_still_assembles_and_reaches_validation() { + let height = 1u64; + let (mut handler, _dir, addr) = make_handler(height); + let peer = test_peer(2); + let block = Block::::new(H256::zero(), height, TestPayload); + let block_bytes = block.encode(); + + let value = run_stream( + &mut handler, + peer, + complete_stream(addr, height, &block_bytes), + ) + .await; + + assert!( + value.is_some(), + "validator peer proposal should be accepted" + ); + let value = value.unwrap(); + assert_eq!(value.height, Height::new(height)); + assert!( + handler + .state + .store + .get_undecided_proposals(Height::new(height), Round::new(0)) + .unwrap() + .iter() + .any(|proposal| proposal.value == value.value), + "accepted validator proposal must reach the existing validation/store path" + ); + } + + #[tokio::test] + async fn validator_peer_allowlist_rotates() { + let current = 10u64; + let (mut handler, _dir, addr) = make_handler(current); + let old_peer = test_peer(2); + let new_peer = test_peer(4); + handler + .state + .update_validator_peers([new_peer].into_iter().collect::>()); + + let old_value = run_stream( + &mut handler, + old_peer, + complete_stream(addr, current + 1, b"old"), + ) + .await; + assert!(old_value.is_none()); + assert_eq!(handler.state.open_proposal_streams(), 0); + + let new_value = run_stream( + &mut handler, + new_peer, + complete_stream(addr, current + 1, b"new"), + ) + .await; + assert!( + new_value.is_none(), + "future-height validator parts are buffered" + ); + let pending = handler + .state + .store + .get_pending_proposal_parts(Height::new(current + 1), Round::new(0)) + .unwrap(); + assert_eq!( + pending.len(), + 1, + "new active validator peer must be accepted after rotation" + ); + } + /// `Externalities` impl whose finalize-side callback always /// fails, so we can drive [`AppMsgHandler::process_finalized`] /// down the fatal path. `process_mb_proposal` must succeed so @@ -1005,6 +1124,7 @@ mod tests { let mut state = State::::new( signer, validator_set, + validator_peers([test_peer(2)]), address, store, Duration::from_secs(1), diff --git a/ethexe/malachite/core/src/config.rs b/ethexe/malachite/core/src/config.rs index f97e716074f..1611aab9971 100644 --- a/ethexe/malachite/core/src/config.rs +++ b/ethexe/malachite/core/src/config.rs @@ -7,19 +7,17 @@ use std::{net::SocketAddr, path::PathBuf, time::Duration}; pub use malachitebft_app_channel::app::net::Multiaddr; -/// One entry of the validator set. The set is fixed for the lifetime -/// of the deployment — to rotate validators every node must be -/// re-bootstrapped from a fresh [`MalachiteConfig`]. -// -// TODO: #5480 add `libp2p_peer_id: PeerId` so receivers can gate -// `ReceivedProposalPart` against a validator-peer-id allowlist -// (libp2p peer-id is not derivable from `public_key` alone — operators -// must compute it offline via `libp2p_peer_id(&secret)` and embed it). +/// One entry of the active validator set. #[derive(Clone, Debug)] pub struct ValidatorEntry { /// secp256k1 public key for this validator. The on-chain address /// is derived from it (`keccak256(uncompressed_pubkey[1..])[12..]`). pub public_key: gsigner::schemes::secp256k1::PublicKey, + /// Libp2p peer ID of this validator's Malachite swarm identity. + /// This is not derivable from [`Self::public_key`] alone because + /// Malachite uses a domain-separated libp2p key. Operators can + /// materialize it offline with [`crate::libp2p_peer_id`]. + pub peer_id: libp2p_identity::PeerId, /// Voting power. Must be > 0; the BFT quorum threshold is /// `> 2/3` of the total voting power across the set. pub voting_power: u64, diff --git a/ethexe/malachite/core/src/service.rs b/ethexe/malachite/core/src/service.rs index 25925b0ae83..56907729f8f 100644 --- a/ethexe/malachite/core/src/service.rs +++ b/ethexe/malachite/core/src/service.rs @@ -6,13 +6,13 @@ use crate::{ app, codec::ScaleCodec, - config::{MalachiteConfig, NodeRole}, + config::{MalachiteConfig, NodeRole, ValidatorEntry}, context::{MalachiteCtx, Validator, ValidatorSet}, externalities::{BlockPayload, Externalities}, signing::{ MalachiteSigner, libp2p_keypair_from, private_key_from_gsigner, public_key_from_gsigner, }, - state::{SharedValidatorSet, State}, + state::{SharedValidatorPeers, SharedValidatorSet, State}, store::Store, types::Address, }; @@ -28,10 +28,12 @@ use malachitebft_app_channel::{ PubSubProtocol, RuntimeConfig, TransportProtocol, ValuePayload, ValueSyncConfig, }, metrics::SharedRegistry, + types::PeerId, }, }; use malachitebft_core_types::ValidatorProof; use std::{ + collections::BTreeSet, marker::PhantomData, pin::Pin, sync::Arc, @@ -52,6 +54,7 @@ pub struct MalachiteService> { /// Shared with the inner app loop; [`Self::update_validators`] /// writes here, the next `Finalized` / `ConsensusReady` reply reads. validator_set: SharedValidatorSet, + validator_peers: SharedValidatorPeers, _externalities: Arc, _phantom: PhantomData P>, } @@ -119,19 +122,15 @@ impl> MalachiteService { let libp2p_keypair = libp2p_keypair_from(&validator_secret_bytes); - // ---- validator set from config ---- + // ---- validator identities from config ---- if config.validators.is_empty() { return Err(anyhow::anyhow!("MalachiteConfig::validators is empty")); } - let mut validators = Vec::with_capacity(config.validators.len()); - for entry in &config.validators { - let pk = public_key_from_gsigner(&entry.public_key) - .context("converting validator public key")?; - validators.push(Validator::new(pk, entry.voting_power)); - } - let initial_validator_set = ValidatorSet::new(validators); + let (initial_validator_set, initial_validator_peers) = + build_validator_state(&config.validators)?; let in_set = initial_validator_set.get_by_address(&address).is_some(); let validator_set = SharedValidatorSet::new(initial_validator_set); + let validator_peers = SharedValidatorPeers::new(initial_validator_peers); // ---- network identity, role-dependent ---- let identity = match config.role { @@ -141,7 +140,26 @@ impl> MalachiteService { "NodeRole::Validator: local address {address} not present in MalachiteConfig::validators" )); } - let peer_id_bytes = libp2p_keypair.public().to_peer_id().to_bytes(); + let local_peer_id = libp2p_keypair.public().to_peer_id(); + let configured_peer_id = config + .validators + .iter() + .find_map(|entry| { + let pk = public_key_from_gsigner(&entry.public_key).ok()?; + (Address::from_public_key(&pk) == address).then_some(entry.peer_id) + }) + .ok_or_else(|| { + anyhow::anyhow!( + "NodeRole::Validator: local address {address} not present in MalachiteConfig::validators" + ) + })?; + if configured_peer_id != local_peer_id { + return Err(anyhow::anyhow!( + "NodeRole::Validator: local validator {address} peer id mismatch: \ + configured {configured_peer_id}, derived {local_peer_id}" + )); + } + let peer_id_bytes = local_peer_id.to_bytes(); // Sign (validator_pubkey, peer_id_bytes) to bind // libp2p identity to the validator's on-chain identity. let signing_provider = MalachiteSigner::new(signer.private_key().clone()); @@ -194,6 +212,7 @@ impl> MalachiteService { let state = State::

::new( signer, validator_set.clone(), + validator_peers.clone(), address, store, config.propose_timeout, @@ -214,6 +233,7 @@ impl> MalachiteService { engine, app_handle, validator_set, + validator_peers, _externalities: externalities, _phantom: PhantomData, }) @@ -234,18 +254,39 @@ impl> MalachiteService { "MalachiteService::update_validators: empty validators list" )); } - let mut converted = Vec::with_capacity(validators.len()); - for entry in &validators { - let pk = public_key_from_gsigner(&entry.public_key) - .context("converting validator public key")?; - converted.push(Validator::new(pk, entry.voting_power)); - } - let new_set = ValidatorSet::new(converted); + let (new_set, new_peers) = build_validator_state(&validators)?; self.validator_set.update(new_set); + self.validator_peers.update(new_peers); Ok(()) } } +fn build_validator_state( + validators: &[ValidatorEntry], +) -> Result<(ValidatorSet, BTreeSet)> { + let mut converted = Vec::with_capacity(validators.len()); + let mut peers = BTreeSet::new(); + for entry in validators { + let pk = public_key_from_gsigner(&entry.public_key) + .with_context(|| format!("converting validator public key {}", entry.public_key))?; + let address = Address::from_public_key(&pk); + let peer = PeerId::from_bytes(&entry.peer_id.to_bytes()).map_err(|e| { + anyhow::anyhow!( + "validator {address} has malformed peer_id {}: {e}", + entry.peer_id + ) + })?; + if !peers.insert(peer) { + return Err(anyhow::anyhow!( + "duplicate Malachite validator peer_id {} for validator {address}", + entry.peer_id + )); + } + converted.push(Validator::new(pk, entry.voting_power)); + } + Ok((ValidatorSet::new(converted), peers)) +} + impl> Stream for MalachiteService { type Item = anyhow::Error; @@ -262,6 +303,99 @@ impl> FusedStream for MalachiteService> MService for MalachiteService {} +#[cfg(test)] +mod tests { + use super::*; + use crate::{ + signing::libp2p_peer_id, + types::{Block, CommitCertificate, H256}, + }; + use async_trait::async_trait; + use gsigner::schemes::secp256k1::PrivateKey; + use parity_scale_codec::{Decode, Encode}; + use std::{sync::Arc, time::Duration}; + + #[derive(Clone, Debug, Encode, Decode, PartialEq, Eq)] + struct TestPayload; + + struct NoopExt; + + #[async_trait] + impl Externalities for NoopExt { + async fn process_mb_proposal(&self, _: H256, _: Block) -> Result<()> { + Ok(()) + } + + async fn process_mb_finalized(&self, _: H256, _: CommitCertificate) -> Result<()> { + Ok(()) + } + + async fn build_block_above(&self, _: H256) -> Result { + Ok(TestPayload) + } + + async fn validate_block_above(&self, _: H256, _: TestPayload) -> Result { + Ok(true) + } + } + + fn secret(seed: u8) -> [u8; 32] { + let mut bytes = [0u8; 32]; + bytes[31] = seed; + bytes + } + + fn validator_entry(seed: u8) -> ValidatorEntry { + let secret = secret(seed); + let private_key = PrivateKey::from_seed(secret).expect("private key"); + ValidatorEntry { + public_key: private_key.public_key(), + peer_id: libp2p_peer_id(&secret), + voting_power: 1, + } + } + + #[test] + fn validator_state_rejects_duplicate_peer_ids() { + let one = validator_entry(1); + let mut two = validator_entry(2); + two.peer_id = one.peer_id; + + let error = build_validator_state(&[one, two]).unwrap_err(); + + assert!( + error + .to_string() + .contains("duplicate Malachite validator peer_id") + ); + } + + #[tokio::test] + async fn validator_start_rejects_local_peer_id_mismatch() { + let local_secret = secret(1); + let local_private_key = PrivateKey::from_seed(local_secret).expect("private key"); + let mut entry = validator_entry(1); + entry.peer_id = libp2p_peer_id(&secret(2)); + let base = tempfile::tempdir().expect("malachite base"); + let config = MalachiteConfig { + listen_addr: "127.0.0.1:0".parse().expect("listen addr"), + base: base.path().to_path_buf(), + persistent_peers: Vec::new(), + validator_secret: local_private_key, + validators: vec![entry], + role: NodeRole::Validator, + propose_timeout: Duration::from_secs(1), + }; + + let error = match MalachiteService::new(config, Arc::new(NoopExt)).await { + Ok(_) => panic!("startup with mismatched peer id must fail"), + Err(error) => error, + }; + + assert!(error.to_string().contains("peer id mismatch")); + } +} + fn build_inner_config(cfg: &MalachiteConfig, moniker: &str) -> InnerNodeConfig { let transport = TransportProtocol::Tcp; let listen_multiaddr = transport.multiaddr( diff --git a/ethexe/malachite/core/src/state.rs b/ethexe/malachite/core/src/state.rs index 3cbe88e0ad6..4a60372a661 100644 --- a/ethexe/malachite/core/src/state.rs +++ b/ethexe/malachite/core/src/state.rs @@ -10,6 +10,7 @@ //! which calls into this struct. use std::{ + collections::BTreeSet, marker::PhantomData, sync::{Arc, RwLock}, time::Duration, @@ -71,9 +72,31 @@ impl SharedValidatorSet { } } +/// Active Malachite peer IDs for proposal-part ingress. +#[derive(Clone)] +pub(crate) struct SharedValidatorPeers(Arc>>); + +impl SharedValidatorPeers { + pub fn new(peers: BTreeSet) -> Self { + Self(Arc::new(RwLock::new(peers))) + } + + pub fn contains(&self, peer: &PeerId) -> bool { + self.0 + .read() + .expect("validator peer set lock poisoned") + .contains(peer) + } + + pub fn update(&self, peers: BTreeSet) { + *self.0.write().expect("validator peer set lock poisoned") = peers; + } +} + pub(crate) struct State { pub signer: MalachiteSigner, pub validator_set: SharedValidatorSet, + validator_peers: SharedValidatorPeers, pub address: Address, pub store: Store

, streams_map: PartStreamsMap, @@ -88,6 +111,7 @@ impl State

{ pub fn new( signer: MalachiteSigner, validator_set: SharedValidatorSet, + validator_peers: SharedValidatorPeers, address: Address, store: Store

, propose_timeout: Duration, @@ -99,6 +123,7 @@ impl State

{ Ok(Self { signer, validator_set, + validator_peers, address, store, streams_map: PartStreamsMap::new(), @@ -114,6 +139,10 @@ impl State

{ self.validator_set.get() } + pub fn is_active_validator_peer(&self, peer: &PeerId) -> bool { + self.validator_peers.contains(peer) + } + /// Round timeouts. Propose phase is bounded by the configured /// [`crate::MalachiteConfig::propose_timeout`] plus a small margin /// for non-proposers; everything else (including the per-round @@ -138,6 +167,16 @@ impl State

{ self.streams_map.insert(from, part) } + #[cfg(test)] + pub fn open_proposal_streams(&self) -> usize { + self.streams_map.len() + } + + #[cfg(test)] + pub fn update_validator_peers(&mut self, peers: BTreeSet) { + self.validator_peers.update(peers); + } + /// Re-assemble a [`ProposedValue`] from a completed /// [`ProposalParts`] sequence. The single `Data` part carries /// the SCALE-encoded block bytes; `Init` supplies the (height, diff --git a/ethexe/malachite/core/src/streaming.rs b/ethexe/malachite/core/src/streaming.rs index 7f9d3c37778..b9bf3ed31bc 100644 --- a/ethexe/malachite/core/src/streaming.rs +++ b/ethexe/malachite/core/src/streaming.rs @@ -186,6 +186,11 @@ impl PartStreamsMap { Self::default() } + #[cfg(test)] + pub fn len(&self) -> usize { + self.streams.len() + } + /// Insert a part. Returns `Some(parts)` once the stream is /// complete (all parts seen + Fin received). Subsequent calls for /// the same `(peer, stream)` after completion return `None` — the diff --git a/ethexe/malachite/core/tests/multi_validators.rs b/ethexe/malachite/core/tests/multi_validators.rs index 2cbc62508fe..4a5d985e5ca 100644 --- a/ethexe/malachite/core/tests/multi_validators.rs +++ b/ethexe/malachite/core/tests/multi_validators.rs @@ -259,6 +259,7 @@ fn validator_entries(setups: &[ValidatorSetup]) -> Vec { .iter() .map(|s| ValidatorEntry { public_key: s.private_key.public_key(), + peer_id: s.peer_id, voting_power: 1, }) .collect() @@ -488,6 +489,7 @@ async fn full_node_syncs_from_validators() { .iter() .map(|s| ValidatorEntry { public_key: s.private_key.public_key(), + peer_id: s.peer_id, voting_power: 1, }) .collect(); diff --git a/ethexe/malachite/service/src/service.rs b/ethexe/malachite/service/src/service.rs index 13c500c22ec..d5e1546569c 100644 --- a/ethexe/malachite/service/src/service.rs +++ b/ethexe/malachite/service/src/service.rs @@ -51,8 +51,8 @@ pub struct MalachiteService { /// by the observer. externalities: Arc, /// On-chain validator addresses only — we keep operator-supplied - /// pub keys here so era rotations can resolve them back. - validator_pool: HashMap, + /// identities here so era rotations can resolve them back. + validator_pool: HashMap, /// Era of the set currently in the engine; gates rotation no-ops. active_era: Option, /// Inner ethexe-malachite-core service. Held in an `Option` so @@ -167,11 +167,11 @@ impl MalachiteService { post_quarantine_delay: config.post_quarantine_delay, }); - // On-chain addresses → pub keys, so era rotations resolve back without an out-of-band lookup. - let validator_pool: HashMap = config + // On-chain addresses → full identities, so era rotations resolve back without an out-of-band lookup. + let validator_pool: HashMap = config .validators .iter() - .map(|v| (v.public_key.to_address(), v.public_key)) + .map(|v| (v.public_key.to_address(), v.clone())) .collect(); let inner = @@ -266,7 +266,7 @@ impl MalachiteService { } /// Push the on-chain validators for `head`'s era into the engine, - /// if the era moved. Skips on missing DB data or unknown pub keys + /// if the era moved. Skips on missing DB data or unknown identities /// (wait-and-retry: the next `BlockSynced` re-evaluates). fn maybe_rotate_validators_for_era(&mut self, head: &SimpleBlockData) { let db = &self.externalities.db; @@ -282,14 +282,11 @@ impl MalachiteService { return; }; - let mut new_set = Vec::with_capacity(self.validator_pool.len()); + let mut new_set = Vec::with_capacity(addrs.len()); let mut missing: Vec

= Vec::new(); for addr in addrs.iter() { match self.validator_pool.get(addr) { - Some(pk) => new_set.push(ValidatorEntry { - public_key: *pk, - voting_power: 1, - }), + Some(identity) => new_set.push(identity.clone()), None => missing.push(*addr), } } @@ -298,7 +295,7 @@ impl MalachiteService { tracing::warn!( era, missing = ?missing, - "validator pool missing pub keys for some on-chain era validators; \ + "validator pool missing identities for some on-chain era validators; \ keeping the previous active set", ); return; diff --git a/ethexe/malachite/service/tests/restart_resilience.rs b/ethexe/malachite/service/tests/restart_resilience.rs index 55170fb8a23..8028635b1e7 100644 --- a/ethexe/malachite/service/tests/restart_resilience.rs +++ b/ethexe/malachite/service/tests/restart_resilience.rs @@ -25,6 +25,7 @@ use ethexe_common::{ use ethexe_db::Database; use ethexe_malachite::{ EmptyMempool, MalachiteConfig, MalachiteEvent, MalachiteService, ValidatorEntry, + malachite_libp2p_peer_id, }; use futures::StreamExt as _; use gprimitives::H256; @@ -90,6 +91,7 @@ fn build_config( home: &Path, listen_port: u16, pub_key: gsigner::schemes::secp256k1::PublicKey, + peer_id: ethexe_malachite::PeerId, ) -> MalachiteConfig { MalachiteConfig { gas_allowance: MalachiteConfig::DEFAULT_GAS_ALLOWANCE, @@ -103,6 +105,7 @@ fn build_config( persistent_peers: Vec::new(), validators: vec![ValidatorEntry { public_key: pub_key, + peer_id, voting_power: 1, }], } @@ -179,10 +182,16 @@ async fn single_validator_finalizes_and_recovers_after_restart() { let chain = seed_chain(&db, 64, 0xDEAD_BEEF); let (signer, pub_key) = build_signer(home.path()); + let peer_id = malachite_libp2p_peer_id( + &signer + .private_key(pub_key) + .expect("validator key") + .to_bytes(), + ); // ---- first run ------------------------------------------------- let mut svc = MalachiteService::new( - build_config(home.path(), 30_001, pub_key), + build_config(home.path(), 30_001, pub_key, peer_id), db.clone(), signer.clone(), Some(pub_key), @@ -224,7 +233,7 @@ async fn single_validator_finalizes_and_recovers_after_restart() { // ---- second run on the SAME home dir + DB ---------------------- let mut svc2 = MalachiteService::new( - build_config(home.path(), 30_001, pub_key), + build_config(home.path(), 30_001, pub_key, peer_id), db.clone(), signer, Some(pub_key), diff --git a/ethexe/scripts/start-local-network.sh b/ethexe/scripts/start-local-network.sh index cd5c7197b35..142d28a373a 100755 --- a/ethexe/scripts/start-local-network.sh +++ b/ethexe/scripts/start-local-network.sh @@ -48,8 +48,8 @@ CONTAINER_PROMETHEUS_PORT="9635" # Malachite BFT consensus uses a separate libp2p TCP swarm (ethexe-network # is QUIC/UDP). Port 20334 is the default malachite listen address; we -# also pre-derive a `validators -> pubkey` JSON file per node (the -# `--validators-malachite-pub-keys` operand). +# also pre-derive one shared `validator -> identity` JSON file and copy it +# into each node directory (the `--validators-malachite-identities` operand). MALACHITE_PORT_START="20334" CONTAINER_MALACHITE_PORT="20334" @@ -680,17 +680,17 @@ generate_keys() { done # Generate one shared `malachite-validators.json` that lists every - # validator's address → public key. Every node loads the same map - # (via `--validators-malachite-pub-keys`) so they all agree on the - # Malachite validator set even though the Router contract only - # stores eth-addresses. + # validator's address → public key + Malachite peer ID. Every node + # loads the same map (via `--validators-malachite-identities`) so + # they all agree on the Malachite validator set even though the + # Router contract only stores eth-addresses. generate_malachite_validators_json } -# Build the validators JSON consumed by `--validators-malachite-pub-keys`. -# Shape: `{ "0x
": "0x", ... }`. The map ordering -# doesn't matter — the service walks the on-chain validator list (in -# router order) and looks each address up here. +# Build the validators JSON consumed by `--validators-malachite-identities`. +# Shape: `{ "0x
": { "public_key": "0x", "peer_id": "" }, ... }`. +# The map ordering doesn't matter — the service walks the on-chain validator +# list (in router order) and looks each address up here. generate_malachite_validators_json() { local json_path="$BASE_DIR/malachite-validators.json" { @@ -698,11 +698,12 @@ generate_malachite_validators_json() { for ((i = 0; i < NUM_VALIDATORS; i++)); do local addr="${VALIDATOR_ADDRESSES[$i]}" local pk="${VALIDATOR_PUB_KEYS[$i]}" + local malachite_peer_id="${MALACHITE_PEER_IDS[$i]}" # Trailing comma on every entry except the last. if [[ $i -lt $((NUM_VALIDATORS - 1)) ]]; then - printf ' "%s": "%s",\n' "$addr" "$pk" + printf ' "%s": { "public_key": "%s", "peer_id": "%s" },\n' "$addr" "$pk" "$malachite_peer_id" else - printf ' "%s": "%s"\n' "$addr" "$pk" + printf ' "%s": { "public_key": "%s", "peer_id": "%s" }\n' "$addr" "$pk" "$malachite_peer_id" fi done echo "}" @@ -759,7 +760,7 @@ start_nodes() { # can advertise the container DNS multiaddr directly. cmd+=" --network-public-addr /dns4/${NODE_CONTAINER_PREFIX}-${i}/udp/$CONTAINER_NETWORK_PORT/quic-v1" cmd+=" --malachite-listen-addr 0.0.0.0:$CONTAINER_MALACHITE_PORT" - cmd+=" --validators-malachite-pub-keys /data/malachite-validators.json" + cmd+=" --validators-malachite-identities /data/malachite-validators.json" if [[ "$ETHEXE_VERBOSE" == "true" ]]; then cmd+=" --verbose" diff --git a/ethexe/service/src/config.rs b/ethexe/service/src/config.rs index a0056db7f38..2a277bdd24e 100644 --- a/ethexe/service/src/config.rs +++ b/ethexe/service/src/config.rs @@ -4,7 +4,7 @@ //! Application config in one place. use anyhow::Result; -use ethexe_malachite::Multiaddr; +use ethexe_malachite::{Multiaddr, PeerId}; use ethexe_network::NetworkConfig; use ethexe_prometheus::PrometheusConfig; use ethexe_rpc::RpcConfig; @@ -35,14 +35,18 @@ pub struct MalachiteCliConfig { /// reachable through the listed ones). pub persistent_peers: Vec, /// Map from validator Ethereum [`Address`] to its Malachite - /// secp256k1 [`PublicKey`]. The on-chain Router contract stores - /// the validator set as Ethereum addresses; Malachite needs the - /// matching public keys to verify votes/proposals. The service - /// resolves the final validator set by walking the on-chain - /// validator list (in router order) and looking each address up - /// in this table, so the table must contain every active - /// validator's address. - pub validator_pub_keys: BTreeMap, + /// signing public key and libp2p peer ID. The on-chain Router + /// contract stores the validator set as Ethereum addresses; + /// Malachite needs the matching public keys to verify votes and + /// the peer IDs to reject proposal parts from non-validator + /// publishers. + pub validator_identities: BTreeMap, +} + +#[derive(Clone, Debug)] +pub struct ValidatorIdentity { + pub public_key: PublicKey, + pub peer_id: PeerId, } impl Default for MalachiteCliConfig { @@ -50,7 +54,7 @@ impl Default for MalachiteCliConfig { Self { listen_addr: ethexe_malachite::MalachiteConfig::DEFAULT_LISTEN_ADDR, persistent_peers: Vec::new(), - validator_pub_keys: BTreeMap::new(), + validator_identities: BTreeMap::new(), } } } diff --git a/ethexe/service/src/lib.rs b/ethexe/service/src/lib.rs index d9ce55f1c88..c6553cde817 100644 --- a/ethexe/service/src/lib.rs +++ b/ethexe/service/src/lib.rs @@ -125,8 +125,8 @@ impl ExternalDataProvider for RouterDataProvider { /// Build the Malachite validator set from the on-chain validator /// list (in router order) by looking each address up in the -/// `address -> public key` table loaded from the -/// `--validators-malachite-pub-keys` JSON file. +/// `address -> validator identity` table loaded from the +/// `--validators-malachite-identities` JSON file. /// /// Voting power is fixed at 1 — Malachite quorum is `> 2/3` of the /// total, which under uniform weights matches the Router's @@ -134,19 +134,20 @@ impl ExternalDataProvider for RouterDataProvider { /// stake, the lookup here is the natural place to plumb it through. fn build_malachite_validator_set( on_chain_validators: impl IntoIterator, - pub_keys: &BTreeMap, + identities: &BTreeMap, ) -> Result> { on_chain_validators .into_iter() .map(|addr| { - let pub_key = pub_keys.get(&addr).copied().with_context(|| { + let identity = identities.get(&addr).with_context(|| { format!( - "validator address {addr} has no entry in --validators-malachite-pub-keys; \ + "validator address {addr} has no entry in --validators-malachite-identities; \ every on-chain validator must be present in the table" ) })?; Ok(ValidatorEntry { - public_key: pub_key, + public_key: identity.public_key, + peer_id: identity.peer_id, voting_power: 1, }) }) @@ -506,7 +507,7 @@ impl Service { let malachite = { let malachite_validator_set = build_malachite_validator_set( validators.iter().copied(), - &config.malachite.validator_pub_keys, + &config.malachite.validator_identities, )?; log::info!( "Malachite validators: {} (local role: {})", @@ -1076,3 +1077,40 @@ impl GenesisInitializer for GenesisInitializerFromFile { .boxed() } } + +#[cfg(test)] +mod malachite_validator_identity_tests { + use super::*; + use ethexe_malachite::malachite_libp2p_peer_id; + + fn identity(seed: u8) -> (Address, config::ValidatorIdentity) { + let mut bytes = [0u8; 32]; + bytes[31] = seed; + let private_key = PrivateKey::from_seed(bytes).expect("private key"); + let public_key = private_key.public_key(); + ( + public_key.to_address(), + config::ValidatorIdentity { + public_key, + peer_id: malachite_libp2p_peer_id(&bytes), + }, + ) + } + + #[test] + fn build_malachite_validator_set_rejects_missing_active_validator_identity() { + let (known_addr, known_identity) = identity(1); + let (missing_addr, _) = identity(2); + let identities = BTreeMap::from([(known_addr, known_identity)]); + + let error = build_malachite_validator_set([known_addr, missing_addr], &identities) + .expect_err("missing identity must fail closed"); + + assert!( + error + .to_string() + .contains("--validators-malachite-identities") + ); + assert!(error.to_string().contains(&missing_addr.to_string())); + } +} diff --git a/ethexe/service/src/tests/utils/env.rs b/ethexe/service/src/tests/utils/env.rs index 7ed110894cc..e08dc23fced 100644 --- a/ethexe/service/src/tests/utils/env.rs +++ b/ethexe/service/src/tests/utils/env.rs @@ -1177,6 +1177,7 @@ impl Node { .iter() .map(|e| ValidatorEntry { public_key: e.pub_key, + peer_id: e.peer_id, voting_power: 1, }) .collect(); diff --git a/ethexe/service/tests/smoke.rs b/ethexe/service/tests/smoke.rs index 1b38c3d826d..7a9cd77b9a0 100644 --- a/ethexe/service/tests/smoke.rs +++ b/ethexe/service/tests/smoke.rs @@ -3,11 +3,12 @@ use ethexe_common::consensus::DEFAULT_BATCH_SIZE_LIMIT; use ethexe_ethereum::{Ethereum, router::RouterQuery}; +use ethexe_malachite::malachite_libp2p_peer_id; use ethexe_prometheus::PrometheusConfig; use ethexe_rpc::{DEFAULT_BLOCK_GAS_LIMIT_MULTIPLIER, RpcConfig}; use ethexe_service::{ Service, - config::{self, Config, EthereumConfig}, + config::{self, Config, EthereumConfig, ValidatorIdentity}, }; use gsigner::secp256k1::Signer; use std::{ @@ -64,24 +65,31 @@ async fn constructor() { // `Service::new` resolves the Malachite validator set by looking // each on-chain validator address up in - // `config.malachite.validator_pub_keys`. The smoke test only + // `config.malachite.validator_identities`. The smoke test only // exercises the constructor wiring (the service is dropped // immediately, nothing signs anything), so populate the table with - // freshly generated keys keyed by the live router's validators. + // freshly generated identities keyed by the live router's validators. let malachite_signer = - Signer::fs(tmp_dir.join("malachite-pub-keys")).expect("failed to create signer"); + Signer::fs(tmp_dir.join("malachite-identities")).expect("failed to create signer"); let router_query = RouterQuery::new(ð_cfg.rpc, eth_cfg.router_address) .await .expect("router query"); let validators = router_query.validators().await.expect("validators"); - let validator_pub_keys = validators + let validator_identities = validators .iter() .map(|addr| { + let public_key = malachite_signer + .generate() + .expect("failed to generate malachite pub key"); + let secret = malachite_signer + .private_key(public_key) + .expect("failed to load malachite private key"); ( *addr, - malachite_signer - .generate() - .expect("failed to generate malachite pub key"), + ValidatorIdentity { + public_key, + peer_id: malachite_libp2p_peer_id(&secret.to_bytes()), + }, ) }) .collect(); @@ -91,7 +99,7 @@ async fn constructor() { ethereum: eth_cfg, network: None, malachite: config::MalachiteCliConfig { - validator_pub_keys, + validator_identities, ..Default::default() }, rpc: None,