verify node instance before accepting

Author: gregcusack

gossip/src/cluster_info.rs

@@ -2505,6 +2505,14 @@ impl ClusterInfo {
                 false
             }
         };
+        let mut verify_node_instance = |value: &CrdsValue| {
+            if self.verify_node_instance(value) {
+                true
+            } else {
+                self.stats.num_unverifed_node_instances.add_relaxed(1);
+                false
+            }
+        };
         // Split packets based on their types.
         let mut pull_requests = vec![];
         let mut pull_responses = vec![];
@@ -2522,13 +2530,15 @@ impl ClusterInfo {
                 Protocol::PullResponse(_, mut data) => {
                     check_duplicate_instance(&data)?;
                     data.retain(&mut verify_gossip_addr);
+                    data.retain(&mut verify_node_instance);
                     if !data.is_empty() {
                         pull_responses.append(&mut data);
                     }
                 }
                 Protocol::PushMessage(from, mut data) => {
                     check_duplicate_instance(&data)?;
                     data.retain(&mut verify_gossip_addr);
+                    data.retain(&mut verify_node_instance);
                     if !data.is_empty() {
                         push_messages.push((from, data));
                     }
@@ -2578,6 +2588,20 @@ impl ClusterInfo {
         Ok(())
     }
 
+    fn verify_node_instance(&self, value: &CrdsValue) -> bool {
+        let pubkey = match &value.data {
+            CrdsData::NodeInstance(node) => node.from(),
+            _ => return true, // If not a NodeInstance, nothing to verify.
+        };
+        // if contact info for the pubkey exists in the crds table, then the
+        // the contact info has already been verified. Therefore, the node
+        // instance is valid.
+        if self.lookup_contact_info(pubkey, |ci| ci.clone()).is_some() {
+            return true;
+        }
+        false
+    }
+
     // Consumes packets received from the socket, deserializing, sanitizing and
     // verifying them and then sending them down the channel for the actual
     // handling of requests/messages.

gossip/src/cluster_info_metrics.rs

@@ -130,6 +130,7 @@ pub struct GossipStats {
     pub(crate) new_push_requests: Counter,
     pub(crate) new_push_requests_num: Counter,
     pub(crate) num_unverifed_gossip_addrs: Counter,
+    pub(crate) num_unverifed_node_instances: Counter,
     pub(crate) packets_received_count: Counter,
     pub(crate) packets_received_ping_messages_count: Counter,
     pub(crate) packets_received_pong_messages_count: Counter,
@@ -494,6 +495,11 @@ pub(crate) fn submit_gossip_stats(
             stats.num_unverifed_gossip_addrs.clear(),
             i64
         ),
+        (
+            "num_unverifed_node_instances",
+            stats.num_unverifed_node_instances.clear(),
+            i64
+        ),
         (
             "packets_received_count",
             stats.packets_received_count.clear(),

gossip/src/crds_value.rs

@@ -455,6 +455,11 @@ impl NodeInstance {
         }
     }
 
+    #[inline]
+    pub fn from(&self) -> &Pubkey {
+        &self.from
+    }
+
     // Clones the value with an updated wallclock.
     pub(crate) fn with_wallclock(&self, wallclock: u64) -> Self {
         Self { wallclock, ..*self }