diff --git a/Cargo.lock b/Cargo.lock index 44348de..4daacb0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6136,6 +6136,7 @@ dependencies = [ "channel", "chrono", "chunk_pool", + "duration-str", "error-chain", "file_location_cache", "futures", @@ -6144,6 +6145,7 @@ dependencies = [ "network", "pruner", "rand 0.8.5", + "serde", "shared_types", "storage", "storage-async", diff --git a/node/router/Cargo.toml b/node/router/Cargo.toml index b87b896..575cf5d 100644 --- a/node/router/Cargo.toml +++ b/node/router/Cargo.toml @@ -21,6 +21,8 @@ chunk_pool = { path = "../chunk_pool" } tokio = { version = "1.19.2", features = ["full"] } tracing = "0.1.35" rand = "0.8.5" +serde = { version = "1.0.137", features = ["derive"] } +duration-str = "0.5.1" [dev-dependencies] channel = { path = "../../common/channel" } diff --git a/node/router/src/lib.rs b/node/router/src/lib.rs index 8b619d8..665b163 100644 --- a/node/router/src/lib.rs +++ b/node/router/src/lib.rs @@ -5,28 +5,42 @@ mod libp2p_event_handler; mod peer_manager; mod service; +use duration_str::deserialize_duration; use network::Multiaddr; +use serde::Deserialize; +use std::time::Duration; pub use crate::service::RouterService; -#[derive(Debug, Clone)] +#[derive(Debug, Clone, Deserialize)] +#[serde(default)] pub struct Config { - pub heartbeat_interval_secs: u64, - - pub idle_time_secs: u64, + #[serde(deserialize_with = "deserialize_duration")] + pub heartbeat_interval: Duration, + #[serde(deserialize_with = "deserialize_duration")] + pub idle_time: Duration, pub max_idle_incoming_peers: usize, pub max_idle_outgoing_peers: usize, pub libp2p_nodes: Vec, + pub private_ip_enabled: bool, } impl Default for Config { fn default() -> Self { Self { - heartbeat_interval_secs: 5, - idle_time_secs: 180, + heartbeat_interval: Duration::from_secs(5), + idle_time: Duration::from_secs(180), max_idle_incoming_peers: 12, max_idle_outgoing_peers: 20, libp2p_nodes: vec![], + private_ip_enabled: false, } } } + +impl Config { + pub fn with_private_ip_enabled(mut self, enabled: bool) -> Self { + self.private_ip_enabled = enabled; + self + } +} diff --git a/node/router/src/libp2p_event_handler.rs b/node/router/src/libp2p_event_handler.rs index 9c52605..01a449c 100644 --- a/node/router/src/libp2p_event_handler.rs +++ b/node/router/src/libp2p_event_handler.rs @@ -2,7 +2,9 @@ use std::{ops::Neg, sync::Arc}; use chunk_pool::ChunkPoolMessage; use file_location_cache::FileLocationCache; +use network::multiaddr::Protocol; use network::types::{AnnounceShardConfig, SignedAnnounceShardConfig}; +use network::Multiaddr; use network::{ rpc::StatusMessage, types::{ @@ -20,6 +22,7 @@ use tokio::sync::mpsc::UnboundedSender; use tokio::sync::{mpsc, RwLock}; use crate::peer_manager::PeerManager; +use crate::Config; lazy_static::lazy_static! { pub static ref FIND_FILE_TIMEOUT: chrono::Duration = chrono::Duration::minutes(2); @@ -65,6 +68,7 @@ fn verify_signature(msg: &dyn HasSignature, peer_id: &PeerId, propagation_source } pub struct Libp2pEventHandler { + config: Config, /// A collection of global variables, accessible outside of the network service. network_globals: Arc, /// A channel to the router service. @@ -86,6 +90,7 @@ pub struct Libp2pEventHandler { impl Libp2pEventHandler { #[allow(clippy::too_many_arguments)] pub fn new( + config: Config, network_globals: Arc, network_send: mpsc::UnboundedSender, sync_send: SyncSender, @@ -96,6 +101,7 @@ impl Libp2pEventHandler { peers: Arc>, ) -> Self { Self { + config, network_globals, network_send, sync_send, @@ -257,16 +263,47 @@ impl Libp2pEventHandler { } } + fn get_listen_addr(&self) -> Option { + let listen_addrs = self.network_globals.listen_multiaddrs.read(); + + if self.config.private_ip_enabled { + listen_addrs.first().cloned() + } else { + listen_addrs + .iter() + .find(|&x| Self::contains_public_ip(x)) + .cloned() + } + } + + fn contains_public_ip(addr: &Multiaddr) -> bool { + for c in addr.iter() { + match c { + Protocol::Ip4(ip4_addr) => { + return !ip4_addr.is_broadcast() + && !ip4_addr.is_documentation() + && !ip4_addr.is_link_local() + && !ip4_addr.is_loopback() + && !ip4_addr.is_multicast() + && !ip4_addr.is_private() + && !ip4_addr.is_unspecified() + } + Protocol::Ip6(ip6_addr) => { + return !ip6_addr.is_loopback() + && !ip6_addr.is_multicast() + && !ip6_addr.is_unspecified() + } + _ => {} + } + } + + false + } + pub async fn construct_announce_file_message(&self, tx_id: TxID) -> Option { let peer_id = *self.network_globals.peer_id.read(); - let addr = match self.network_globals.listen_multiaddrs.read().first() { - Some(addr) => addr.clone(), - None => { - error!("No listen address available"); - return None; - } - }; + let addr = self.get_listen_addr()?; let timestamp = timestamp_now(); let shard_config = self.store.get_store().flow().get_shard_config(); @@ -298,13 +335,7 @@ impl Libp2pEventHandler { shard_config: ShardConfig, ) -> Option { let peer_id = *self.network_globals.peer_id.read(); - let addr = match self.network_globals.listen_multiaddrs.read().first() { - Some(addr) => addr.clone(), - None => { - error!("No listen address available"); - return None; - } - }; + let addr = self.get_listen_addr()?; let timestamp = timestamp_now(); let msg = AnnounceShardConfig { @@ -377,12 +408,7 @@ impl Libp2pEventHandler { index_end: u64, ) -> Option { let peer_id = *self.network_globals.peer_id.read(); - let addr = self - .network_globals - .listen_multiaddrs - .read() - .first()? - .clone(); + let addr = self.get_listen_addr()?; let timestamp = timestamp_now(); let msg = AnnounceChunks { @@ -473,6 +499,12 @@ impl Libp2pEventHandler { return MessageAcceptance::Reject; } + // verify public ip address if required + let addr = msg.at.clone().into(); + if !self.config.private_ip_enabled && !Self::contains_public_ip(&addr) { + return MessageAcceptance::Reject; + } + // propagate gossip to peers let d = duration_since(msg.resend_timestamp); if d < TOLERABLE_DRIFT.neg() || d > *ANNOUNCE_FILE_TIMEOUT { @@ -484,7 +516,7 @@ impl Libp2pEventHandler { self.send_to_sync(SyncMessage::AnnounceFileGossip { tx_id: msg.tx_id, peer_id: msg.peer_id.clone().into(), - addr: msg.at.clone().into(), + addr, }); // insert message to cache @@ -503,6 +535,12 @@ impl Libp2pEventHandler { return MessageAcceptance::Reject; } + // verify public ip address if required + let addr = msg.at.clone().into(); + if !self.config.private_ip_enabled && !Self::contains_public_ip(&addr) { + return MessageAcceptance::Reject; + } + // propagate gossip to peers let d = duration_since(msg.resend_timestamp); if d < TOLERABLE_DRIFT.neg() || d > *ANNOUNCE_SHARD_CONFIG_TIMEOUT { @@ -518,7 +556,7 @@ impl Libp2pEventHandler { self.send_to_sync(SyncMessage::AnnounceShardConfig { shard_config, peer_id: msg.peer_id.clone().into(), - addr: msg.at.clone().into(), + addr, }); // insert message to cache @@ -538,6 +576,12 @@ impl Libp2pEventHandler { return MessageAcceptance::Reject; } + // verify public ip address if required + let addr = msg.at.clone().into(); + if !self.config.private_ip_enabled && !Self::contains_public_ip(&addr) { + return MessageAcceptance::Reject; + } + // propagate gossip to peers let d = duration_since(msg.resend_timestamp); if d < TOLERABLE_DRIFT.neg() || d > *ANNOUNCE_FILE_TIMEOUT { @@ -625,6 +669,7 @@ mod tests { impl Context { fn new_handler(&self) -> Libp2pEventHandler { Libp2pEventHandler::new( + Config::default().with_private_ip_enabled(true), self.network_globals.clone(), self.network_send.clone(), self.sync_send.clone(), diff --git a/node/router/src/peer_manager.rs b/node/router/src/peer_manager.rs index 243e3ac..6d1bc6e 100644 --- a/node/router/src/peer_manager.rs +++ b/node/router/src/peer_manager.rs @@ -2,7 +2,7 @@ use crate::Config; use network::PeerId; use rand::seq::IteratorRandom; use std::collections::HashMap; -use std::time::Instant; +use std::time::{Duration, Instant}; /// Connected peer info. struct PeerInfo { @@ -20,8 +20,8 @@ impl PeerInfo { } } - fn elapsed_secs(&self) -> u64 { - self.since.elapsed().as_secs() + fn elapsed(&self) -> Duration { + self.since.elapsed() } } @@ -98,7 +98,7 @@ impl PeerManager { .peers .iter() .filter(|(_, peer)| { - peer.outgoing == outgoing && peer.elapsed_secs() >= self.config.idle_time_secs + peer.outgoing == outgoing && peer.elapsed() >= self.config.idle_time }) .map(|(peer_id, _)| *peer_id) .collect(); @@ -116,10 +116,7 @@ impl PeerManager { #[cfg(test)] mod tests { - use std::{ - ops::Sub, - time::{Duration, Instant}, - }; + use std::{ops::Sub, time::Instant}; use network::PeerId; @@ -202,10 +199,9 @@ mod tests { } // change timestamp for all peers - let idle_timeout = Duration::from_secs(config.idle_time_secs); for peer_id in peers.iter() { let peer = manager.peers.get_mut(peer_id).unwrap(); - peer.since = Instant::now().sub(idle_timeout); + peer.since = Instant::now().sub(config.idle_time); } assert_eq!( diff --git a/node/router/src/service.rs b/node/router/src/service.rs index 36cbbb5..2399758 100644 --- a/node/router/src/service.rs +++ b/node/router/src/service.rs @@ -10,7 +10,6 @@ use network::{ }; use pruner::PrunerMessage; use std::sync::Arc; -use std::time::Duration; use storage::log_store::Store as LogStore; use storage_async::Store; use sync::{SyncMessage, SyncSender}; @@ -68,13 +67,14 @@ impl RouterService { // create the network service and spawn the task let router = RouterService { - config, + config: config.clone(), libp2p, network_globals: network_globals.clone(), network_recv, pruner_recv, peers: peers.clone(), libp2p_event_handler: Libp2pEventHandler::new( + config, network_globals, network_send, sync_send, @@ -94,7 +94,7 @@ impl RouterService { } async fn main(mut self, mut shutdown_sender: Sender) { - let mut heartbeat = interval(Duration::from_secs(self.config.heartbeat_interval_secs)); + let mut heartbeat = interval(self.config.heartbeat_interval); loop { tokio::select! { @@ -194,6 +194,7 @@ impl RouterService { } }, Libp2pEvent::NewListenAddr(multiaddr) => { + info!(?multiaddr, "New listen address"); self.network_globals .listen_multiaddrs .write() diff --git a/node/src/config/convert.rs b/node/src/config/convert.rs index 724f51a..8659ad3 100644 --- a/node/src/config/convert.rs +++ b/node/src/config/convert.rs @@ -180,7 +180,7 @@ impl ZgsConfig { } pub fn router_config(&self, network_config: &NetworkConfig) -> Result { - let mut router_config = router::Config::default(); + let mut router_config = self.router.clone(); router_config.libp2p_nodes = network_config.libp2p_nodes.to_vec(); Ok(router_config) } diff --git a/node/src/config/mod.rs b/node/src/config/mod.rs index 7364a17..6512de2 100644 --- a/node/src/config/mod.rs +++ b/node/src/config/mod.rs @@ -87,6 +87,9 @@ build_config! { pub struct ZgsConfig { pub raw_conf: RawConfiguration, + // router config, configured by [router] section by `config` crate. + pub router: router::Config, + // sync config, configured by [sync] section by `config` crate. pub sync: sync::Config, } diff --git a/tests/config/node_config.py b/tests/config/node_config.py index 1c41724..192b954 100644 --- a/tests/config/node_config.py +++ b/tests/config/node_config.py @@ -3,6 +3,9 @@ from web3 import Web3 ZGS_CONFIG = { "log_config_file": "log_config", "confirmation_block_count": 1, + "router": { + "private_ip_enabled": True, + } } BSC_CONFIG = dict(