Refuse network identity incompatible nodes in UDP discovery layer (#253)

* Add python test for UDP discovery

* Refuse nodes with incompatible ENR

node/network/src/behaviour/mod.rs

@@ -267,7 +267,7 @@ impl<AppReqId: ReqId> Behaviour<AppReqId> {
             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);

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,
         }
     }
 }

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_ext::{CombinedKeyExt, ENR_CONTENT_KEY_NETWORK_ID};
-use super::ENR_FILENAME;
+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<T: EnrKey>(
         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<Enr, String> {
     let enr_f = dir.join(ENR_FILENAME);

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<Multiaddr>;
+
+    /// Returns network identity in content.
+    fn network_identity(&self) -> Option<Result<NetworkIdentity, ssz::DecodeError>>;
 }
 
 /// 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<Result<NetworkIdentity, ssz::DecodeError>> {
+        let value = self.get(ENR_CONTENT_KEY_NETWORK_ID)?;
+        Some(NetworkIdentity::from_ssz_bytes(value))
+    }
 }
 
 impl CombinedKeyPublicExt for CombinedPublicKey {

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,

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,

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<NetworkGlobals>,
     ) -> error::Result<Self> {
         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,

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::<Result<_, _>>()
             .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)
     }

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,
 

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,

tests/config/zgs/network/enr.dat

@@ -0,0 +1 @@
+enr:-Ly4QJZwz9htAorBIx_otqoaRFPohX7NQJ31iBB6mcEhBiuPWsOnigc1ABQsg6tLU1OirQdLR6aEvv8SlkkfIbV72T8CgmlkgnY0gmlwhH8AAAGQbmV0d29ya19pZGVudGl0eZ8oIwAAAAAAADPyz8cpvYcPpUtQMmYOBrTPKn-UAAIAiXNlY3AyNTZrMaEDeDdgnDgLPkxNxB39jKb9f1Na30t6R9vVolpTk5zu-hODdGNwgir4g3VkcIIq-A

tests/config/zgs/network/key

@@ -0,0 +1 @@
+Y<13><><02><><EFBFBD>Ң<>-
<0A><>r<>7<EFBFBD><37>jq<6A>p<>}<7D>

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()

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()

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<dyn Fn(&Enr) -> 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 {
+    pub fn table_filter<F>(&mut self, filter: F) -> &mut Self where F: Fn(&Enr) -> bool + Send + Sync + 'static {
-        self.config.table_filter = filter;
+        self.config.table_filter = Arc::new(filter);
         self
     }
 

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.

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