adds hash domain to ping-pong protocol (backport #27193) (#27243)

* adds hash domain to ping-pong protocol (#27193)

In order to maintain backward compatibility, for now the responding node
will hash the token both with and without domain so that the other node
will accept the response regardless of its upgrade status.
Once the cluster has upgraded to the new code, we will remove the legacy
domain = false case.

(cherry picked from commit 6928b2a)

# Conflicts:
#	gossip/src/cluster_info.rs

* removes mergify merge conflicts

Co-authored-by: behzad nouri <[email protected]>

core/src/ancestor_hashes_service.rs

@@ -425,16 +425,21 @@ impl AncestorHashesService {
                     stats.invalid_packets += 1;
                     return None;
                 }
-                if ping.verify() {
-                    stats.ping_count += 1;
-                    if let Ok(pong) = Pong::new(&ping, keypair) {
+                if !ping.verify() {
+                    stats.ping_err_verify_count += 1;
+                    return None;
+                }
+                stats.ping_count += 1;
+                // Respond both with and without domain so that the other node
+                // will accept the response regardless of its upgrade status.
+                // TODO: remove domain = false once cluster is upgraded.
+                for domain in [false, true] {
+                    if let Ok(pong) = Pong::new(domain, &ping, keypair) {
                         let pong = RepairProtocol::Pong(pong);
                         if let Ok(pong_bytes) = serialize(&pong) {
                             let _ignore = ancestor_socket.send_to(&pong_bytes[..], from_addr);
                         }
                     }
-                } else {
-                    stats.ping_err_verify_count += 1;
                 }
                 None
             }

core/src/serve_repair.rs

@@ -1031,11 +1031,16 @@ impl ServeRepair {
                 }
                 packet.meta.set_discard(true);
                 stats.ping_count += 1;
-                if let Ok(pong) = Pong::new(&ping, keypair) {
-                    let pong = RepairProtocol::Pong(pong);
-                    if let Ok(pong_bytes) = serialize(&pong) {
-                        let from_addr = packet.meta.socket_addr();
-                        pending_pongs.push((pong_bytes, from_addr));
+                // Respond both with and without domain so that the other node
+                // will accept the response regardless of its upgrade status.
+                // TODO: remove domain = false once cluster is upgraded.
+                for domain in [false, true] {
+                    if let Ok(pong) = Pong::new(domain, &ping, keypair) {
+                        let pong = RepairProtocol::Pong(pong);
+                        if let Ok(pong_bytes) = serialize(&pong) {
+                            let from_addr = packet.meta.socket_addr();
+                            pending_pongs.push((pong_bytes, from_addr));
+                        }
                     }
                 }
             }

gossip/src/cluster_info.rs

@@ -2160,20 +2160,23 @@ impl ClusterInfo {
         I: IntoIterator<Item = (SocketAddr, Ping)>,
     {
         let keypair = self.keypair();
-        let packets: Vec<_> = pings
-            .into_iter()
-            .filter_map(|(addr, ping)| {
-                let pong = Pong::new(&ping, &keypair).ok()?;
-                let pong = Protocol::PongMessage(pong);
-                match Packet::from_data(Some(&addr), pong) {
-                    Ok(packet) => Some(packet),
-                    Err(err) => {
-                        error!("failed to write pong packet: {:?}", err);
-                        None
+        let mut packets = Vec::new();
+        for (addr, ping) in pings {
+            // Respond both with and without domain so that the other node will
+            // accept the response regardless of its upgrade status.
+            // TODO: remove domain = false once cluster is upgraded.
+            for domain in [false, true] {
+                if let Ok(pong) = Pong::new(domain, &ping, &keypair) {
+                    let pong = Protocol::PongMessage(pong);
+                    match Packet::from_data(Some(&addr), pong) {
+                        Ok(packet) => packets.push(packet),
+                        Err(err) => {
+                            error!("failed to write pong packet: {:?}", err);
+                        }
                     }
                 }
-            })
-            .collect();
+            }
+        }
         if packets.is_empty() {
             None
         } else {
@@ -3181,7 +3184,9 @@ mod tests {
         let pongs: Vec<(SocketAddr, Pong)> = pings
             .iter()
             .zip(&remote_nodes)
-            .map(|(ping, (keypair, socket))| (*socket, Pong::new(ping, keypair).unwrap()))
+            .map(|(ping, (keypair, socket))| {
+                (*socket, Pong::new(/*domain:*/ true, ping, keypair).unwrap())
+            })
             .collect();
         let now = now + Duration::from_millis(1);
         cluster_info.handle_batch_pong_messages(pongs, now);
@@ -3224,7 +3229,7 @@ mod tests {
             .collect();
         let pongs: Vec<_> = pings
             .iter()
-            .map(|ping| Pong::new(ping, &this_node).unwrap())
+            .map(|ping| Pong::new(/*domain:*/ false, ping, &this_node).unwrap())
             .collect();
         let recycler = PacketBatchRecycler::default();
         let packets = cluster_info
@@ -3236,9 +3241,9 @@ mod tests {
                 &recycler,
             )
             .unwrap();
-        assert_eq!(remote_nodes.len(), packets.len());
+        assert_eq!(remote_nodes.len() * 2, packets.len());
         for (packet, (_, socket), pong) in izip!(
-            packets.into_iter(),
+            packets.into_iter().step_by(2),
             remote_nodes.into_iter(),
             pongs.into_iter()
         ) {

gossip/src/ping_pong.rs

@@ -16,6 +16,8 @@ use {
     },
 };
 
+const PING_PONG_HASH_PREFIX: &[u8] = "SOLANA_PING_PONG".as_bytes();
+
 #[derive(AbiExample, Debug, Deserialize, Serialize)]
 pub struct Ping<T> {
     from: Pubkey,
@@ -100,8 +102,17 @@ impl<T: Serialize> Signable for Ping<T> {
 }
 
 impl Pong {
-    pub fn new<T: Serialize>(ping: &Ping<T>, keypair: &Keypair) -> Result<Self, Error> {
-        let hash = hash::hash(&serialize(&ping.token)?);
+    pub fn new<T: Serialize>(
+        domain: bool,
+        ping: &Ping<T>,
+        keypair: &Keypair,
+    ) -> Result<Self, Error> {
+        let token = serialize(&ping.token)?;
+        let hash = if domain {
+            hash::hashv(&[PING_PONG_HASH_PREFIX, &token])
+        } else {
+            hash::hash(&token)
+        };
         let pong = Pong {
             from: keypair.pubkey(),
             hash,
@@ -187,9 +198,15 @@ impl PingCache {
             Some(t) if now.saturating_duration_since(*t) < delay => None,
             _ => {
                 let ping = pingf()?;
-                let hash = hash::hash(&serialize(&ping.token).ok()?);
-                self.pings.put(node, now);
+                let token = serialize(&ping.token).ok()?;
+                // For backward compatibility, for now responses both with and
+                // without domain are accepted.
+                // TODO: remove no domain case once cluster is upgraded.
+                let hash = hash::hash(&token);
+                self.pending_cache.put(hash, node);
+                let hash = hash::hashv(&[PING_PONG_HASH_PREFIX, &token]);
                 self.pending_cache.put(hash, node);
+                self.pings.put(node, now);
                 Some(ping)
             }
         }
@@ -281,10 +298,18 @@ mod tests {
         assert!(ping.verify());
         assert!(ping.sanitize().is_ok());
 
-        let pong = Pong::new(&ping, &keypair).unwrap();
+        let pong = Pong::new(/*domain:*/ false, &ping, &keypair).unwrap();
         assert!(pong.verify());
         assert!(pong.sanitize().is_ok());
         assert_eq!(hash::hash(&ping.token), pong.hash);
+
+        let pong = Pong::new(/*domian:*/ true, &ping, &keypair).unwrap();
+        assert!(pong.verify());
+        assert!(pong.sanitize().is_ok());
+        assert_eq!(
+            hash::hashv(&[PING_PONG_HASH_PREFIX, &ping.token]),
+            pong.hash
+        );
     }
 
     #[test]
@@ -339,7 +364,10 @@ mod tests {
                     assert!(ping.is_none());
                 }
                 Some(ping) => {
-                    let pong = Pong::new(ping, keypair).unwrap();
+                    let domain = rng.gen_ratio(1, 2);
+                    let pong = Pong::new(domain, ping, keypair).unwrap();
+                    assert!(cache.add(&pong, *socket, now));
+                    let pong = Pong::new(!domain, ping, keypair).unwrap();
                     assert!(cache.add(&pong, *socket, now));
                 }
             }