diff --git a/node/network/src/behaviour/mod.rs b/node/network/src/behaviour/mod.rs index fb562ac..e848e26 100644 --- a/node/network/src/behaviour/mod.rs +++ b/node/network/src/behaviour/mod.rs @@ -267,7 +267,7 @@ impl Behaviour { discovery_enabled: !config.disable_discovery, metrics_enabled: config.metrics_enabled, target_peer_count: config.target_peers, - ..Default::default() + ..config.peer_manager }; let slot_duration = std::time::Duration::from_secs(12); diff --git a/node/network/src/config.rs b/node/network/src/config.rs index 8778052..e7fbd21 100644 --- a/node/network/src/config.rs +++ b/node/network/src/config.rs @@ -1,6 +1,5 @@ -use crate::peer_manager::peerdb::PeerDBConfig; use crate::types::GossipKind; -use crate::{Enr, PeerIdSerialized}; +use crate::{peer_manager, Enr, PeerIdSerialized}; use directory::{ DEFAULT_BEACON_NODE_DIR, DEFAULT_HARDCODED_NETWORK, DEFAULT_NETWORK_DIR, DEFAULT_ROOT_DIR, }; @@ -128,7 +127,12 @@ pub struct Config { /// The id of the storage network. pub network_id: NetworkIdentity, - pub peer_db: PeerDBConfig, + pub peer_db: peer_manager::peerdb::PeerDBConfig, + pub peer_manager: peer_manager::config::Config, + + /// Whether to disable network identity in ENR. + /// This is for test purpose only. + pub disable_enr_network_id: bool, } impl Default for Config { @@ -208,6 +212,8 @@ impl Default for Config { metrics_enabled: false, network_id: Default::default(), peer_db: Default::default(), + peer_manager: Default::default(), + disable_enr_network_id: false, } } } diff --git a/node/network/src/discovery/enr.rs b/node/network/src/discovery/enr.rs index a9dfde7..45d099d 100644 --- a/node/network/src/discovery/enr.rs +++ b/node/network/src/discovery/enr.rs @@ -1,9 +1,10 @@ //! Helper functions and an extension trait for Ethereum 2 ENRs. pub use discv5::enr::{CombinedKey, EnrBuilder}; +use ssz::Encode; -use super::enr_ext::CombinedKeyExt; -use super::ENR_FILENAME; +use super::enr_ext::{CombinedKeyExt, ENR_CONTENT_KEY_NETWORK_ID}; +use super::{EnrExt, ENR_FILENAME}; use crate::types::Enr; use crate::NetworkConfig; use discv5::enr::EnrKey; @@ -32,7 +33,9 @@ pub fn use_or_load_enr( Ok(disk_enr) => { // if the same node id, then we may need to update our sequence number if local_enr.node_id() == disk_enr.node_id() { - if compare_enr(local_enr, &disk_enr) { + if compare_enr(local_enr, &disk_enr) + && is_disk_enr_network_id_unchanged(&disk_enr, config) + { debug!(file = ?enr_f, "ENR loaded from disk"); // the stored ENR has the same configuration, use it *local_enr = disk_enr; @@ -94,6 +97,13 @@ pub fn create_enr_builder_from_config( let tcp_port = config.enr_tcp_port.unwrap_or(config.libp2p_port); builder.tcp(tcp_port); } + // add network identity info in ENR if not disabled + if !config.disable_enr_network_id { + builder.add_value( + ENR_CONTENT_KEY_NETWORK_ID, + &config.network_id.as_ssz_bytes(), + ); + } builder } @@ -117,6 +127,14 @@ fn compare_enr(local_enr: &Enr, disk_enr: &Enr) -> bool { && (local_enr.udp().is_none() || local_enr.udp() == disk_enr.udp()) } +fn is_disk_enr_network_id_unchanged(disk_enr: &Enr, config: &NetworkConfig) -> bool { + match disk_enr.network_identity() { + Some(Ok(id)) => !config.disable_enr_network_id && id == config.network_id, + Some(Err(_)) => false, + None => config.disable_enr_network_id, + } +} + /// Loads enr from the given directory pub fn load_enr_from_disk(dir: &Path) -> Result { let enr_f = dir.join(ENR_FILENAME); diff --git a/node/network/src/discovery/enr_ext.rs b/node/network/src/discovery/enr_ext.rs index 81eaaaf..d483270 100644 --- a/node/network/src/discovery/enr_ext.rs +++ b/node/network/src/discovery/enr_ext.rs @@ -2,8 +2,12 @@ use crate::{Enr, Multiaddr, PeerId}; use discv5::enr::{CombinedKey, CombinedPublicKey}; use libp2p::core::{identity::Keypair, identity::PublicKey, multiaddr::Protocol}; +use shared_types::NetworkIdentity; +use ssz::Decode; use tiny_keccak::{Hasher, Keccak}; +pub(crate) const ENR_CONTENT_KEY_NETWORK_ID: &'static str = "network_identity"; + /// Extend ENR for libp2p types. pub trait EnrExt { /// The libp2p `PeerId` for the record. @@ -24,6 +28,9 @@ pub trait EnrExt { /// Returns any multiaddrs that contain the TCP protocol. fn multiaddr_tcp(&self) -> Vec; + + /// Returns network identity in content. + fn network_identity(&self) -> Option>; } /// Extend ENR CombinedPublicKey for libp2p types. @@ -189,6 +196,12 @@ impl EnrExt for Enr { } multiaddrs } + + /// Returns network identity in content. + fn network_identity(&self) -> Option> { + let value = self.get(ENR_CONTENT_KEY_NETWORK_ID)?; + Some(NetworkIdentity::from_ssz_bytes(value)) + } } impl CombinedKeyPublicExt for CombinedPublicKey { diff --git a/node/network/src/discovery/mod.rs b/node/network/src/discovery/mod.rs index 307377c..554b64e 100644 --- a/node/network/src/discovery/mod.rs +++ b/node/network/src/discovery/mod.rs @@ -139,6 +139,7 @@ impl Discovery { udp = ?local_enr.udp(), tcp = ?local_enr.tcp(), udp4_socket = ?local_enr.udp_socket(), + network_id = ?local_enr.network_identity(), "ENR Initialised", ); @@ -158,6 +159,7 @@ impl Discovery { ip = ?bootnode_enr.ip(), udp = ?bootnode_enr.udp(), tcp = ?bootnode_enr.tcp(), + network_id = ?bootnode_enr.network_identity(), "Adding node to routing table", ); let repr = bootnode_enr.to_string(); @@ -205,13 +207,37 @@ impl Discovery { match result { Ok(enr) => { debug!( + multiaddr = %original_addr.to_string(), node_id = %enr.node_id(), peer_id = %enr.peer_id(), ip = ?enr.ip(), udp = ?enr.udp(), tcp = ?enr.tcp(), - "Adding node to routing table", + network_id = ?enr.network_identity(), + "Adding bootnode to routing table", ); + + // check network identity in bootnode ENR if required + if !config.disable_enr_network_id { + match enr.network_identity() { + Some(Ok(id)) => { + if id != config.network_id { + error!(bootnode=?id, local=?config.network_id, "Bootnode network identity mismatch"); + continue; + } + } + Some(Err(err)) => { + error!(?err, "Failed to decode bootnode network identity"); + continue; + } + None => { + error!("Bootnode has no network identity"); + continue; + } + } + } + + // add bootnode into routing table let _ = discv5.add_enr(enr).map_err(|e| { error!( addr = %original_addr.to_string(), @@ -401,10 +427,16 @@ impl Discovery { // Generate a random target node id. let random_node = NodeId::random(); + // only discover nodes with same network identity + let local_network_id = self.network_globals.network_id(); + let predicate = move |enr: &Enr| -> bool { + matches!(enr.network_identity(), Some(Ok(id)) if id == local_network_id) + }; + // Build the future let query_future = self .discv5 - .find_node_predicate(random_node, Box::new(|_| true), target_peers) + .find_node_predicate(random_node, Box::new(predicate), target_peers) .map(|v| QueryResult { query_type: query, result: v, diff --git a/node/network/src/peer_manager/config.rs b/node/network/src/peer_manager/config.rs index 6c5523d..5520601 100644 --- a/node/network/src/peer_manager/config.rs +++ b/node/network/src/peer_manager/config.rs @@ -1,3 +1,8 @@ +use std::time::Duration; + +use duration_str::deserialize_duration; +use serde::{Deserialize, Serialize}; + /// The time in seconds between re-status's peers. pub const DEFAULT_STATUS_INTERVAL: u64 = 300; @@ -11,9 +16,14 @@ pub const DEFAULT_PING_INTERVAL_INBOUND: u64 = 20; pub const DEFAULT_TARGET_PEERS: usize = 50; /// Configurations for the PeerManager. -#[derive(Debug)] +#[derive(Debug, Clone, Copy, Serialize, Deserialize)] +#[serde(default)] pub struct Config { /* Peer count related configurations */ + /// The heartbeat performs regular updates such as updating reputations and performing discovery + /// requests. This defines the interval in seconds. + #[serde(deserialize_with = "deserialize_duration")] + pub heartbeat_interval: Duration, /// Whether discovery is enabled. pub discovery_enabled: bool, /// Whether metrics are enabled. @@ -35,6 +45,7 @@ pub struct Config { impl Default for Config { fn default() -> Self { Config { + heartbeat_interval: Duration::from_secs(30), discovery_enabled: true, metrics_enabled: false, target_peer_count: DEFAULT_TARGET_PEERS, diff --git a/node/network/src/peer_manager/mod.rs b/node/network/src/peer_manager/mod.rs index e9b33d0..9ee4936 100644 --- a/node/network/src/peer_manager/mod.rs +++ b/node/network/src/peer_manager/mod.rs @@ -30,10 +30,6 @@ use std::net::IpAddr; pub mod config; mod network_behaviour; -/// The heartbeat performs regular updates such as updating reputations and performing discovery -/// requests. This defines the interval in seconds. -const HEARTBEAT_INTERVAL: u64 = 30; - /// This is used in the pruning logic. We avoid pruning peers on sync-committees if doing so would /// lower our peer count below this number. Instead we favour a non-uniform distribution of subnet /// peers. @@ -105,6 +101,7 @@ impl PeerManager { network_globals: Arc, ) -> error::Result { let config::Config { + heartbeat_interval, discovery_enabled, metrics_enabled, target_peer_count, @@ -114,7 +111,7 @@ impl PeerManager { } = cfg; // Set up the peer manager heartbeat interval - let heartbeat = tokio::time::interval(tokio::time::Duration::from_secs(HEARTBEAT_INTERVAL)); + let heartbeat = tokio::time::interval(heartbeat_interval); Ok(PeerManager { network_globals, diff --git a/node/src/config/convert.rs b/node/src/config/convert.rs index 437b348..aae214f 100644 --- a/node/src/config/convert.rs +++ b/node/src/config/convert.rs @@ -5,10 +5,11 @@ use ethereum_types::{H256, U256}; use ethers::prelude::{Http, Middleware, Provider}; use log_entry_sync::{CacheConfig, ContractAddress, LogSyncConfig}; use miner::MinerConfig; -use network::NetworkConfig; +use network::{EnrExt, NetworkConfig}; use pruner::PrunerConfig; use shared_types::{NetworkIdentity, ProtocolVersion}; use std::net::IpAddr; +use std::sync::Arc; use std::time::Duration; use storage::config::ShardConfig; use storage::log_store::log_manager::LogConfig; @@ -38,7 +39,7 @@ impl ZgsConfig { .await .map_err(|e| format!("Unable to get chain id: {:?}", e))? .as_u64(); - network_config.network_id = NetworkIdentity { + let local_network_id = NetworkIdentity { chain_id, flow_address, p2p_protocol_version: ProtocolVersion { @@ -47,6 +48,7 @@ impl ZgsConfig { build: network::PROTOCOL_VERSION[2], }, }; + network_config.network_id = local_network_id.clone(); if !self.network_disable_discovery { network_config.enr_tcp_port = Some(self.network_enr_tcp_port); @@ -82,7 +84,13 @@ impl ZgsConfig { .collect::>() .map_err(|e| format!("Unable to parse network_libp2p_nodes: {:?}", e))?; - network_config.discv5_config.table_filter = |_| true; + network_config.discv5_config.table_filter = if self.discv5_disable_enr_network_id { + Arc::new(|_| true) + } else { + Arc::new( + move |enr| matches!(enr.network_identity(), Some(Ok(id)) if id == local_network_id), + ) + }; network_config.discv5_config.request_timeout = Duration::from_secs(self.discv5_request_timeout_secs); network_config.discv5_config.query_peer_timeout = @@ -97,6 +105,8 @@ impl ZgsConfig { network_config.private = self.network_private; network_config.peer_db = self.network_peer_db; + network_config.peer_manager = self.network_peer_manager; + network_config.disable_enr_network_id = self.discv5_disable_enr_network_id; Ok(network_config) } diff --git a/node/src/config/mod.rs b/node/src/config/mod.rs index 6d30745..0ed1267 100644 --- a/node/src/config/mod.rs +++ b/node/src/config/mod.rs @@ -28,6 +28,7 @@ build_config! { (discv5_report_discovered_peers, (bool), false) (discv5_disable_packet_filter, (bool), false) (discv5_disable_ip_limit, (bool), false) + (discv5_disable_enr_network_id, (bool), false) // log sync (blockchain_rpc_endpoint, (String), "http://127.0.0.1:8545".to_string()) @@ -87,6 +88,9 @@ pub struct ZgsConfig { /// Network peer db config, configured by [network_peer_db] section by `config` crate. pub network_peer_db: network::peer_manager::peerdb::PeerDBConfig, + /// Network peer manager config, configured by [network_peer_manager] section by `config` crate. + pub network_peer_manager: network::peer_manager::config::Config, + // router config, configured by [router] section by `config` crate. pub router: router::Config, diff --git a/tests/config/node_config.py b/tests/config/node_config.py index 35f0754..bcb42c9 100644 --- a/tests/config/node_config.py +++ b/tests/config/node_config.py @@ -3,6 +3,10 @@ from web3 import Web3 ZGS_CONFIG = { "log_config_file": "log_config", "confirmation_block_count": 1, + "discv5_disable_ip_limit": True, + "network_peer_manager": { + "heartbeat_interval": "1s" + }, "router": { "private_ip_enabled": True, }, @@ -18,6 +22,8 @@ ZGS_CONFIG = { } } +ZGS_NODEID = "16Uiu2HAmLkGFUbNFYdhuSbTQ5hmnPjFXx2zUDtwQ2uihHpN9YNNe" + BSC_CONFIG = dict( NetworkId=1000, HTTPPort=8545, diff --git a/tests/config/zgs/network/enr.dat b/tests/config/zgs/network/enr.dat new file mode 100644 index 0000000..dc87bde --- /dev/null +++ b/tests/config/zgs/network/enr.dat @@ -0,0 +1 @@ +enr:-Ly4QJZwz9htAorBIx_otqoaRFPohX7NQJ31iBB6mcEhBiuPWsOnigc1ABQsg6tLU1OirQdLR6aEvv8SlkkfIbV72T8CgmlkgnY0gmlwhH8AAAGQbmV0d29ya19pZGVudGl0eZ8oIwAAAAAAADPyz8cpvYcPpUtQMmYOBrTPKn-UAAIAiXNlY3AyNTZrMaEDeDdgnDgLPkxNxB39jKb9f1Na30t6R9vVolpTk5zu-hODdGNwgir4g3VkcIIq-A \ No newline at end of file diff --git a/tests/config/zgs/network/key b/tests/config/zgs/network/key new file mode 100644 index 0000000..0fa80bd --- /dev/null +++ b/tests/config/zgs/network/key @@ -0,0 +1 @@ +YҢ- r7jqp} \ No newline at end of file diff --git a/tests/network_discovery_test.py b/tests/network_discovery_test.py new file mode 100644 index 0000000..aa06f76 --- /dev/null +++ b/tests/network_discovery_test.py @@ -0,0 +1,74 @@ +#!/usr/bin/env python3 + +import os +import time + +from config.node_config import ZGS_NODEID +from test_framework.test_framework import TestFramework +from utility.utils import p2p_port + +class NetworkDiscoveryTest(TestFramework): + """ + This is to test whether community nodes could connect to each other via UDP discovery. + """ + + def setup_params(self): + # 1 bootnode and 2 community nodes + self.num_nodes = 3 + + # setup for node 0 as bootnode + tests_dir = os.path.dirname(__file__) + network_dir = os.path.join(tests_dir, "config", "zgs", "network") + bootnode_port = p2p_port(0) + self.zgs_node_configs[0] = { + # load pre-defined keypair + "network_dir": network_dir, + + # enable UDP discovery relevant configs + "network_enr_address": "127.0.0.1", + "network_enr_tcp_port": bootnode_port, + "network_enr_udp_port": bootnode_port, + + # disable trusted nodes + "network_libp2p_nodes": [], + } + + # setup node 1 & 2 as community nodes + bootnodes = [f"/ip4/127.0.0.1/udp/{bootnode_port}/p2p/{ZGS_NODEID}"] + for i in range(1, self.num_nodes): + self.zgs_node_configs[i] = { + # enable UDP discovery relevant configs + "network_enr_address": "127.0.0.1", + "network_enr_tcp_port": p2p_port(i), + "network_enr_udp_port": p2p_port(i), + + # disable trusted nodes and enable bootnodes + "network_libp2p_nodes": [], + "network_boot_nodes": bootnodes, + } + + def run_test(self): + timeout_secs = 10 + + for iter in range(timeout_secs + 1): + assert iter < timeout_secs, "Timeout to discover nodes for peer connection" + time.sleep(1) + self.log.info("==================================== iter %s", iter) + + total_connected = 0 + for i in range(self.num_nodes): + info = self.nodes[i].rpc.admin_getNetworkInfo() + total_connected += info["connectedPeers"] + self.log.info( + "Node[%s] peers: total = %s, banned = %s, disconnected = %s, connected = %s (in = %s, out = %s)", + i, info["totalPeers"], info["bannedPeers"], info["disconnectedPeers"], info["connectedPeers"], info["connectedIncomingPeers"], info["connectedOutgoingPeers"], + ) + + if total_connected >= self.num_nodes * (self.num_nodes - 1): + break + + self.log.info("====================================") + self.log.info("All nodes connected to each other successfully") + +if __name__ == "__main__": + NetworkDiscoveryTest().main() diff --git a/tests/network_discovery_upgrade_test.py b/tests/network_discovery_upgrade_test.py new file mode 100644 index 0000000..d37ce1e --- /dev/null +++ b/tests/network_discovery_upgrade_test.py @@ -0,0 +1,74 @@ +#!/usr/bin/env python3 + +import os +import time + +from config.node_config import ZGS_NODEID +from test_framework.test_framework import TestFramework +from utility.utils import p2p_port + +class NetworkDiscoveryUpgradeTest(TestFramework): + """ + This is to test that low version community nodes could not connect to bootnodes. + """ + + def setup_params(self): + # 1 bootnode and 1 community node + self.num_nodes = 2 + + # setup for node 0 as bootnode + tests_dir = os.path.dirname(__file__) + network_dir = os.path.join(tests_dir, "config", "zgs", "network") + bootnode_port = p2p_port(0) + self.zgs_node_configs[0] = { + # load pre-defined keypair + "network_dir": network_dir, + + # enable UDP discovery relevant configs + "network_enr_address": "127.0.0.1", + "network_enr_tcp_port": bootnode_port, + "network_enr_udp_port": bootnode_port, + + # disable trusted nodes + "network_libp2p_nodes": [], + } + + # setup node 1 as community node + bootnodes = [f"/ip4/127.0.0.1/udp/{bootnode_port}/p2p/{ZGS_NODEID}"] + for i in range(1, self.num_nodes): + self.zgs_node_configs[i] = { + # enable UDP discovery relevant configs + "network_enr_address": "127.0.0.1", + "network_enr_tcp_port": p2p_port(i), + "network_enr_udp_port": p2p_port(i), + + # disable trusted nodes and enable bootnodes + "network_libp2p_nodes": [], + "network_boot_nodes": bootnodes, + + # disable network identity in ENR + "discv5_disable_enr_network_id": True, + } + + def run_test(self): + for iter in range(10): + time.sleep(1) + self.log.info("==================================== iter %s", iter) + + total_connected = 0 + for i in range(self.num_nodes): + info = self.nodes[i].rpc.admin_getNetworkInfo() + total_connected += info["connectedPeers"] + self.log.info( + "Node[%s] peers: total = %s, banned = %s, disconnected = %s, connected = %s (in = %s, out = %s)", + i, info["totalPeers"], info["bannedPeers"], info["disconnectedPeers"], info["connectedPeers"], info["connectedIncomingPeers"], info["connectedOutgoingPeers"], + ) + + # ENR incompatible and should not discover each other for TCP connection + assert total_connected == 0, "Nodes connected unexpectedly" + + self.log.info("====================================") + self.log.info("ENR incompatible nodes do not connect to each other") + +if __name__ == "__main__": + NetworkDiscoveryUpgradeTest().main() diff --git a/version-meld/discv5/src/config.rs b/version-meld/discv5/src/config.rs index 264e885..b89b668 100644 --- a/version-meld/discv5/src/config.rs +++ b/version-meld/discv5/src/config.rs @@ -2,7 +2,7 @@ use crate::{ kbucket::MAX_NODES_PER_BUCKET, Enr, Executor, PermitBanList, RateLimiter, RateLimiterBuilder, }; ///! A set of configuration parameters to tune the discovery protocol. -use std::time::Duration; +use std::{sync::Arc, time::Duration}; /// Configuration parameters that define the performance of the gossipsub network. #[derive(Clone)] @@ -57,7 +57,7 @@ pub struct Discv5Config { /// A filter used to decide whether to insert nodes into our local routing table. Nodes can be /// excluded if they do not pass this filter. The default is to accept all nodes. - pub table_filter: fn(&Enr) -> bool, + pub table_filter: Arc bool + Send + Sync>, /// The time between pings to ensure connectivity amongst connected nodes. Default: 300 /// seconds. @@ -123,7 +123,7 @@ impl Default for Discv5Config { query_parallelism: 3, ip_limit: false, incoming_bucket_limit: MAX_NODES_PER_BUCKET, - table_filter: |_| true, + table_filter: Arc::new(|_| true), ping_interval: Duration::from_secs(300), report_discovered_peers: true, filter_rate_limiter, @@ -242,8 +242,8 @@ impl Discv5ConfigBuilder { /// A filter used to decide whether to insert nodes into our local routing table. Nodes can be /// excluded if they do not pass this filter. - pub fn table_filter(&mut self, filter: fn(&Enr) -> bool) -> &mut Self { - self.config.table_filter = filter; + pub fn table_filter(&mut self, filter: F) -> &mut Self where F: Fn(&Enr) -> bool + Send + Sync + 'static { + self.config.table_filter = Arc::new(filter); self } diff --git a/version-meld/discv5/src/handler/mod.rs b/version-meld/discv5/src/handler/mod.rs index 2aebd31..f1e00db 100644 --- a/version-meld/discv5/src/handler/mod.rs +++ b/version-meld/discv5/src/handler/mod.rs @@ -754,6 +754,7 @@ impl Handler { // failed. enr.node_id() == node_address.node_id && (enr.udp_socket().is_none() || enr.udp_socket() == Some(node_address.socket_addr)) + && enr.get("network_identity").is_some() } /// Handle a message that contains an authentication header. diff --git a/version-meld/discv5/src/service.rs b/version-meld/discv5/src/service.rs index 7d5d6e0..8d9542a 100644 --- a/version-meld/discv5/src/service.rs +++ b/version-meld/discv5/src/service.rs @@ -1264,7 +1264,12 @@ impl Service { "Session established with Node: {}, direction: {}", node_id, direction ); - self.connection_updated(node_id, ConnectionStatus::Connected(enr, direction)); + + // requires network identity in ENR, so as to refuse low version peers. + match enr.get("network_identity") { + Some(_) => self.connection_updated(node_id, ConnectionStatus::Connected(enr, direction)), + None => debug!(ip=?enr.ip(), "No network identity in peer ENR"), + } } /// A session could not be established or an RPC request timed-out (after a few retries, if