Filter global traffic (#123)

We are soon going to allow public nodes to follow, however we do not
want their network traffic to interfere with the voting. Public nodes
should be able to consume and subscribe but not publish.

PR allows internal nodes to filter traffic based on a whitelist i.e they
will not accept fragments that are not in the whitelist.

This allows us to configure and deploy multiple scenarios including a
gateway model.

Author: minikin

src/jormungandr/jormungandr-lib/src/interfaces/config/node.rs

@@ -211,6 +211,8 @@ pub struct P2p {
 
     pub allow_private_addresses: bool,
 
+    pub whitelist: Option<Vec<SocketAddr>>,
+
     pub policy: Option<Policy>,
 
     #[serde(skip_serializing_if = "Option::is_none")]

src/jormungandr/jormungandr/src/fragment/process.rs

@@ -114,7 +114,6 @@ impl Process {
                 tip,
                 stats_counter.clone()
             );
-
             loop {
                 tokio::select! {
                     maybe_msg = input.next() => {

src/jormungandr/jormungandr/src/lib.rs

@@ -291,6 +291,7 @@ fn start_services(bootstrapped_node: BootstrappedNode) -> Result<(), start_up::E
 
     {
         let blockchain_tip = blockchain_tip.clone();
+
         let process = fragment::Process::new(
             bootstrapped_node.settings.mempool.pool_max_entries.into(),
             bootstrapped_node.settings.mempool.log_max_entries.into(),

src/jormungandr/jormungandr/src/network/p2p/comm.rs

@@ -666,6 +666,11 @@ impl Peers {
         }
     }
 
+    pub async fn get_peer_addr(&self, peer: &NodeId) -> Option<SocketAddr> {
+        let mut map = self.inner().await;
+        map.peer_comms(peer).map(|peer| peer.remote_addr())
+    }
+
     pub async fn refresh_peer_on_gossip(&self, peer: &NodeId) -> bool {
         let timestamp = SystemTime::now();
         let mut map = self.inner().await;

src/jormungandr/jormungandr/src/network/p2p/comm/peer_map.rs

@@ -267,7 +267,7 @@ impl PeerMap {
             .iter()
             .map(|(&id, data)| PeerInfo {
                 id,
-                addr: None,
+                addr: Some(data.comms.remote_addr),
                 stats: data.stats.clone(),
             })
             .collect()

src/jormungandr/jormungandr/src/network/subscription.rs

@@ -10,6 +10,7 @@ use chain_network::{
     data as net_data,
     error::{Code, Error},
 };
+use futures::executor;
 use futures::{future::BoxFuture, prelude::*, ready};
 use jormungandr_lib::interfaces::FragmentOrigin;
 use std::{
@@ -24,7 +25,7 @@ fn filter_gossip_node(node: &Gossip, config: &Configuration) -> bool {
     if config.allow_private_addresses {
         node.has_valid_address()
     } else {
-        node.is_global()
+        !node.is_global()
     }
 }
 
@@ -165,6 +166,12 @@ impl FragmentProcessor {
         }
     }
 
+    fn get_ingress_addr(&self) -> Option<std::net::SocketAddr> {
+        let state = self.global_state.clone();
+        let node_id = self.node_id;
+        executor::block_on(state.peers.get_peer_addr(&node_id))
+    }
+
     fn refresh_stat(&mut self) {
         let state = self.global_state.clone();
         let node_id = self.node_id;
@@ -301,7 +308,23 @@ impl Sink<net_data::Fragment> for FragmentProcessor {
             e
         })?;
         tracing::debug!(hash = %fragment.hash(), "received fragment");
-        self.buffered_fragments.push(fragment);
+
+        if let Some(whitelist) = &self.global_state.config.whitelist {
+            match self.get_ingress_addr() {
+                Some(ingress_addr) => {
+                    if whitelist.contains(&ingress_addr) {
+                        self.buffered_fragments.push(fragment);
+                    } else {
+                        tracing::info!("dropping fragments from {}", ingress_addr);
+                    }
+                }
+                None => tracing::warn!("unable to resolve address of ingress client"),
+            }
+        } else {
+            // if no whitelist config, normal behaviour, no filtering
+            self.buffered_fragments.push(fragment);
+        }
+
         Ok(())
     }
 
@@ -410,6 +433,7 @@ impl Sink<net_data::Gossip> for GossipProcessor {
         let state1 = self.global_state.clone();
         let mut mbox = self.mbox.clone();
         let node_id = self.node_id;
+
         let fut = future::join(
             async move {
                 let refreshed = state1.peers.refresh_peer_on_gossip(&node_id).await;

src/jormungandr/jormungandr/src/settings/start/config.rs

@@ -10,7 +10,7 @@ pub use jormungandr_lib::interfaces::{Cors, JRpc, LayersConfig, Rest, Tls, Trust
 use jormungandr_lib::{interfaces::Mempool, time::Duration};
 use multiaddr::Multiaddr;
 use serde::{de::Error as _, Deserialize, Deserializer, Serialize, Serializer};
-use std::path::PathBuf;
+use std::{net::SocketAddr, path::PathBuf};
 use tracing::level_filters::LevelFilter;
 
 #[derive(Debug, Deserialize)]
@@ -99,6 +99,8 @@ pub struct P2pConfig {
     #[serde(default)]
     pub allow_private_addresses: bool,
 
+    pub whitelist: Option<Vec<SocketAddr>>,
+
     /// setting for the policy
     #[serde(default)]
     pub policy: QuarantineConfig,

src/jormungandr/jormungandr/src/settings/start/mod.rs

@@ -328,6 +328,7 @@ fn generate_network(
             .unwrap_or(network::DEFAULT_MAX_CLIENT_CONNECTIONS),
         timeout: std::time::Duration::from_secs(15),
         allow_private_addresses: p2p.allow_private_addresses,
+        whitelist: p2p.whitelist,
         gossip_interval: p2p
             .gossip_interval
             .map(|d| d.into())

src/jormungandr/jormungandr/src/settings/start/network.rs

@@ -85,6 +85,8 @@ pub struct Configuration {
     /// Whether to allow non-public IP addresses in gossip
     pub allow_private_addresses: bool,
 
+    pub whitelist: Option<Vec<SocketAddr>>,
+
     pub gossip_interval: Duration,
 
     pub network_stuck_check: Duration,

src/jormungandr/testing/hersir/src/config/spawn_params.rs

@@ -19,6 +19,7 @@ use std::{
 #[derive(Clone, Debug, Deserialize)]
 pub struct SpawnParams {
     alias: NodeAlias,
+
     bootstrap_from_peers: Option<bool>,
     faketime: Option<FaketimeConfig>,
     gossip_interval: Option<Duration>,
@@ -28,6 +29,8 @@ pub struct SpawnParams {
     log_level: Option<LogLevel>,
     max_bootstrap_attempts: Option<usize>,
     max_connections: Option<u32>,
+    allow_private_addresses: Option<bool>,
+    whitelist: Option<Vec<SocketAddr>>,
     max_inbound_connections: Option<u32>,
     mempool: Option<Mempool>,
     network_stuck_check: Option<Duration>,
@@ -59,6 +62,8 @@ impl SpawnParams {
             log_level: None,
             max_bootstrap_attempts: None,
             max_connections: None,
+            allow_private_addresses: None,
+            whitelist: None,
             max_inbound_connections: None,
             mempool: None,
             network_stuck_check: None,
@@ -138,6 +143,16 @@ impl SpawnParams {
         self
     }
 
+    pub fn allow_private_addresses(mut self, switch: bool) -> Self {
+        self.allow_private_addresses = Some(switch);
+        self
+    }
+
+    pub fn whitelist(mut self, nodes: Vec<SocketAddr>) -> Self {
+        self.whitelist = Some(nodes);
+        self
+    }
+
     pub fn max_inbound_connections(mut self, max_inbound_connections: u32) -> Self {
         self.max_inbound_connections = Some(max_inbound_connections);
         self
@@ -273,6 +288,14 @@ impl SpawnParams {
             node_config.p2p.max_inbound_connections = Some(*max_inbound_connections);
         }
 
+        if let Some(allow_private_addresses) = &self.allow_private_addresses {
+            node_config.p2p.allow_private_addresses = *allow_private_addresses;
+        }
+
+        if let Some(whitelist) = &self.whitelist {
+            node_config.p2p.whitelist = Some(whitelist.clone());
+        }
+
         if let Some(max_connections) = &self.max_connections {
             node_config.p2p.max_connections = Some(*max_connections);
         }