Merge pull request #3163 from eqlabs/feat/nopeergossip

feat: allow-external-peers

.devnet/start.sh

@@ -37,7 +37,7 @@ start_snarkos_in_tmux() {
     tmux new-session -d -s snarkos-session
 
     # Send the snarkOS start command to the tmux session with the NODE_ID
-    tmux send-keys -t "snarkos-session" "snarkos start --nodisplay --bft 0.0.0.0:5000 --rest 0.0.0.0:3030 --peers $NODE_IP:4130 --validators $NODE_IP:5000 --rest-rps 1000 --verbosity $VERBOSITY --dev $NODE_ID --dev-num-validators $NUM_INSTANCES --validator --metrics" C-m
+    tmux send-keys -t "snarkos-session" "snarkos start --nodisplay --bft 0.0.0.0:5000 --rest 0.0.0.0:3030 --allow-external-peers --peers $NODE_IP:4130 --validators $NODE_IP:5000 --rest-rps 1000 --verbosity $VERBOSITY --dev $NODE_ID --dev-num-validators $NUM_INSTANCES --validator --metrics" C-m
 
     exit  # Exit root user
 EOF

cli/src/commands/start.rs

@@ -145,9 +145,13 @@ pub struct Start {
     #[clap(long = "storage_path")]
     pub storage_path: Option<PathBuf>,
 
-    #[clap(long)]
     /// If development mode is enabled, specify the custom bonded balances as a json object. (default: None)
+    #[clap(long)]
     dev_bonded_balances: Option<BondedBalances>,
+
+    /// If the flag is set, the validator will allow untrusted peers to connect
+    #[clap(long = "allow-external-peers")]
+    allow_external_peers: bool,
 }
 
 impl Start {
@@ -545,7 +549,7 @@ impl Start {
         // Initialize the node.
         let bft_ip = if self.dev.is_some() { self.bft } else { None };
         match node_type {
-            NodeType::Validator => Node::new_validator(self.node, bft_ip, rest_ip, self.rest_rps, account, &trusted_peers, &trusted_validators, genesis, cdn, storage_mode, dev_txs).await,
+            NodeType::Validator => Node::new_validator(self.node, bft_ip, rest_ip, self.rest_rps, account, &trusted_peers, &trusted_validators, genesis, cdn, storage_mode, self.allow_external_peers, dev_txs).await,
             NodeType::Prover => Node::new_prover(self.node, account, &trusted_peers, genesis, storage_mode).await,
             NodeType::Client => Node::new_client(self.node, rest_ip, self.rest_rps, account, &trusted_peers, genesis, cdn, storage_mode).await,
         }

devnet.sh

@@ -64,12 +64,12 @@ for validator_index in "${validator_indices[@]}"; do
 
   # Send the command to start the validator to the new window and capture output to the log file
   if [ "$validator_index" -eq 0 ]; then
-    tmux send-keys -t "devnet:window$validator_index" "snarkos start --nodisplay --dev $validator_index --dev-num-validators $total_validators --validator --logfile $log_file --metrics" C-m
+    tmux send-keys -t "devnet:window$validator_index" "snarkos start --nodisplay --dev $validator_index --allow-external-peers --dev-num-validators $total_validators --validator --logfile $log_file --metrics" C-m
   else
     # Create a new window with a unique name
     window_index=$((validator_index + index_offset))
     tmux new-window -t "devnet:$window_index" -n "window$validator_index"
-    tmux send-keys -t "devnet:window$validator_index" "snarkos start --nodisplay --dev $validator_index --dev-num-validators $total_validators --validator --logfile $log_file" C-m
+    tmux send-keys -t "devnet:window$validator_index" "snarkos start --nodisplay --dev $validator_index --allow-external-peers --dev-num-validators $total_validators --validator --logfile $log_file" C-m
   fi
 done
 

node/router/src/handshake.rs

@@ -264,6 +264,10 @@ impl<N: Network> Router<N> {
         if self.is_connected(&peer_ip) {
             bail!("Dropping connection request from '{peer_ip}' (already connected)")
         }
+        // Only allow trusted peers to connect if allow_external_peers is set
+        if !self.allow_external_peers() && !self.is_trusted(&peer_ip) {
+            bail!("Dropping connection request from '{peer_ip}' (untrusted)")
+        }
         // Ensure the peer is not restricted.
         if self.is_restricted(&peer_ip) {
             bail!("Dropping connection request from '{peer_ip}' (restricted)")

node/router/src/heartbeat.rs

@@ -107,6 +107,11 @@ pub trait Heartbeat<N: Network>: Outbound<N> {
             return;
         }
 
+        // Skip if the node is not requesting peers.
+        if !self.router().allow_external_peers() {
+            return;
+        }
+
         // Retrieve the trusted peers.
         let trusted = self.router().trusted_peers();
         // Retrieve the bootstrap peers.
@@ -216,9 +221,11 @@ pub trait Heartbeat<N: Network>: Outbound<N> {
             for peer_ip in self.router().candidate_peers().into_iter().choose_multiple(rng, num_deficient) {
                 self.router().connect(peer_ip);
             }
-            // Request more peers from the connected peers.
-            for peer_ip in self.router().connected_peers().into_iter().choose_multiple(rng, 3) {
-                self.send(peer_ip, Message::PeerRequest(PeerRequest));
+            if self.router().allow_external_peers() {
+                // Request more peers from the connected peers.
+                for peer_ip in self.router().connected_peers().into_iter().choose_multiple(rng, 3) {
+                    self.send(peer_ip, Message::PeerRequest(PeerRequest));
+                }
             }
         }
     }

node/router/src/inbound.rs

@@ -125,6 +125,9 @@ pub trait Inbound<N: Network>: Reading + Outbound<N> {
                 if !self.router().cache.contains_outbound_peer_request(peer_ip) {
                     bail!("Peer '{peer_ip}' is not following the protocol (unexpected peer response)")
                 }
+                if !self.router().allow_external_peers() {
+                    bail!("Not accepting peer response from '{peer_ip}' (validator gossip is disabled)");
+                }
 
                 match self.peer_response(peer_ip, &message.peers) {
                     true => Ok(()),

node/router/src/lib.rs

@@ -93,6 +93,8 @@ pub struct InnerRouter<N: Network> {
     restricted_peers: RwLock<HashMap<SocketAddr, Instant>>,
     /// The spawned handles.
     handles: Mutex<Vec<JoinHandle<()>>>,
+    /// If the flag is set, the node will engage in P2P gossip to request more peers.
+    allow_external_peers: bool,
     /// The boolean flag for the development mode.
     is_dev: bool,
 }
@@ -115,6 +117,7 @@ impl<N: Network> Router<N> {
         account: Account<N>,
         trusted_peers: &[SocketAddr],
         max_peers: u16,
+        allow_external_peers: bool,
         is_dev: bool,
     ) -> Result<Self> {
         // Initialize the TCP stack.
@@ -132,6 +135,7 @@ impl<N: Network> Router<N> {
             candidate_peers: Default::default(),
             restricted_peers: Default::default(),
             handles: Default::default(),
+            allow_external_peers,
             is_dev,
         })))
     }
@@ -251,6 +255,11 @@ impl<N: Network> Router<N> {
         self.is_dev
     }
 
+    /// Returns `true` if the node is engaging in P2P gossip to request more peers.
+    pub fn allow_external_peers(&self) -> bool {
+        self.allow_external_peers
+    }
+
     /// Returns the listener IP address from the (ambiguous) peer address.
     pub fn resolve_to_listener(&self, peer_addr: &SocketAddr) -> Option<SocketAddr> {
         self.resolver.get_listener(peer_addr)
@@ -295,6 +304,11 @@ impl<N: Network> Router<N> {
             .unwrap_or(false)
     }
 
+    /// Returns `true` if the given IP is trusted.
+    pub fn is_trusted(&self, ip: &SocketAddr) -> bool {
+        self.trusted_peers.contains(ip)
+    }
+
     /// Returns the maximum number of connected peers.
     pub fn max_connected_peers(&self) -> usize {
         self.tcp.config().max_connections as usize

node/router/tests/common/mod.rs

@@ -78,6 +78,7 @@ pub async fn client(listening_port: u16, max_peers: u16) -> TestRouter<CurrentNe
         &[],
         max_peers,
         true,
+        true,
     )
     .await
     .expect("couldn't create client router")
@@ -94,6 +95,7 @@ pub async fn prover(listening_port: u16, max_peers: u16) -> TestRouter<CurrentNe
         &[],
         max_peers,
         true,
+        true,
     )
     .await
     .expect("couldn't create prover router")
@@ -109,6 +111,7 @@ pub async fn validator(listening_port: u16, max_peers: u16) -> TestRouter<Curren
         sample_account(),
         &[],
         max_peers,
+        false,
         true,
     )
     .await

node/src/client/mod.rs

@@ -108,6 +108,8 @@ impl<N: Network, C: ConsensusStorage<N>> Client<N, C> {
         let ledger_service = Arc::new(CoreLedgerService::<N, C>::new(ledger.clone(), shutdown.clone()));
         // Initialize the sync module.
         let sync = BlockSync::new(BlockSyncMode::Router, ledger_service.clone());
+        // Determine if the client should allow external peers.
+        let allow_external_peers = true;
 
         // Initialize the node router.
         let router = Router::new(
@@ -116,6 +118,7 @@ impl<N: Network, C: ConsensusStorage<N>> Client<N, C> {
             account,
             trusted_peers,
             Self::MAXIMUM_NUMBER_OF_PEERS as u16,
+            allow_external_peers,
             matches!(storage_mode, StorageMode::Development(_)),
         )
         .await?;

node/src/node.rs

@@ -50,6 +50,7 @@ impl<N: Network> Node<N> {
         genesis: Block<N>,
         cdn: Option<String>,
         storage_mode: StorageMode,
+        allow_external_peers: bool,
         dev_txs: bool,
     ) -> Result<Self> {
         Ok(Self::Validator(Arc::new(
@@ -64,6 +65,7 @@ impl<N: Network> Node<N> {
                 genesis,
                 cdn,
                 storage_mode,
+                allow_external_peers,
                 dev_txs,
             )
             .await?,

node/src/prover/mod.rs

@@ -101,6 +101,8 @@ impl<N: Network, C: ConsensusStorage<N>> Prover<N, C> {
         let ledger_service = Arc::new(ProverLedgerService::new());
         // Initialize the sync module.
         let sync = BlockSync::new(BlockSyncMode::Router, ledger_service.clone());
+        // Determine if the prover should allow external peers.
+        let allow_external_peers = true;
 
         // Initialize the node router.
         let router = Router::new(
@@ -109,6 +111,7 @@ impl<N: Network, C: ConsensusStorage<N>> Prover<N, C> {
             account,
             trusted_peers,
             Self::MAXIMUM_NUMBER_OF_PEERS as u16,
+            allow_external_peers,
             matches!(storage_mode, StorageMode::Development(_)),
         )
         .await?;

node/src/validator/mod.rs

@@ -83,6 +83,7 @@ impl<N: Network, C: ConsensusStorage<N>> Validator<N, C> {
         genesis: Block<N>,
         cdn: Option<String>,
         storage_mode: StorageMode,
+        allow_external_peers: bool,
         dev_txs: bool,
     ) -> Result<Self> {
         // Prepare the shutdown flag.
@@ -125,6 +126,7 @@ impl<N: Network, C: ConsensusStorage<N>> Validator<N, C> {
             account,
             trusted_peers,
             Self::MAXIMUM_NUMBER_OF_PEERS as u16,
+            allow_external_peers,
             matches!(storage_mode, StorageMode::Development(_)),
         )
         .await?;
@@ -498,6 +500,7 @@ mod tests {
             genesis,
             None,
             storage_mode,
+            false,
             dev_txs,
         )
         .await

node/tests/common/node.rs

@@ -59,6 +59,7 @@ pub async fn validator() -> Validator<CurrentNetwork, ConsensusMemory<CurrentNet
         sample_genesis_block(), // Should load the current network's genesis block.
         None,                   // No CDN.
         StorageMode::Production,
+        true,  // This test requires validators to connect to peers.
         false, // No dev traffic in production mode.
     )
     .await