allow multicast addresses for TVU

Author: alexpyattaev

gossip/src/contact_info.rs

@@ -329,7 +329,7 @@ impl ContactInfo {
                             num_addrs: self.addrs.len(),
                         })?;
                 let socket = SocketAddr::new(*addr, port);
-                sanitize_socket(&socket)?;
+                sanitize_socket(&socket, key)?;
                 return Ok(socket);
             }
         }
@@ -349,7 +349,11 @@ impl ContactInfo {
     }
 
     pub fn set_socket(&mut self, key: u8, socket: SocketAddr) -> Result<(), Error> {
-        sanitize_socket(&socket)?;
+        if key == SOCKET_TAG_TVU {
+            sanitize_turbine_socket(&socket)?
+        } else {
+            sanitize_socket(&socket, key)?;
+        }
         // Remove the old entry associated with this key (if any).
         self.remove_socket(key);
         // Find the index at which the new socket entry would be inserted into
@@ -463,7 +467,7 @@ impl ContactInfo {
     // Only for tests and simulations.
     pub fn new_with_socketaddr(pubkey: &Pubkey, socket: &SocketAddr) -> Self {
         use Protocol::{QUIC, UDP};
-        assert_matches!(sanitize_socket(socket), Ok(()));
+        assert_matches!(sanitize_socket(socket, SOCKET_TAG_GOSSIP), Ok(()));
         let mut node = Self::new(
             *pubkey,
             solana_time_utils::timestamp(), // wallclock,
@@ -576,7 +580,7 @@ impl TryFrom<ContactInfoLite> for ContactInfo {
                 continue;
             };
             let socket = SocketAddr::new(addr, port);
-            if sanitize_socket(&socket).is_ok() {
+            if sanitize_socket(&socket, key).is_ok() {
                 *entry = socket;
             }
         }
@@ -593,20 +597,31 @@ impl Sanitize for ContactInfo {
     }
 }
 
-pub(crate) fn sanitize_socket(socket: &SocketAddr) -> Result<(), Error> {
+pub(crate) fn sanitize_socket(socket: &SocketAddr, key: u8) -> Result<(), Error> {
     if socket.port() == 0u16 {
         return Err(Error::InvalidPort(socket.port()));
     }
     let addr = socket.ip();
     if addr.is_unspecified() {
         return Err(Error::UnspecifiedIpAddr(addr));
     }
-    if addr.is_multicast() {
+    if key != SOCKET_TAG_TVU && addr.is_multicast() {
         return Err(Error::MulticastIpAddr(addr));
     }
     Ok(())
 }
 
+pub(crate) fn sanitize_turbine_socket(socket: &SocketAddr) -> Result<(), Error> {
+    if socket.port() == 0u16 {
+        return Err(Error::InvalidPort(socket.port()));
+    }
+    let addr = socket.ip();
+    if addr.is_unspecified() {
+        return Err(Error::UnspecifiedIpAddr(addr));
+    }
+    Ok(())
+}
+
 // Sanitizes deserialized IpAddr and socket entries.
 fn sanitize_entries(addrs: &[IpAddr], sockets: &[SocketEntry]) -> Result<(), Error> {
     // Verify that all IP addresses are unique.
@@ -863,7 +878,7 @@ mod tests {
             let addr = addrs.choose(&mut rng).unwrap();
             let socket = SocketAddr::new(*addr, new_rand_port(&mut rng));
             let key = rng.gen_range(KEYS.start..KEYS.end);
-            if sanitize_socket(&socket).is_ok() {
+            if sanitize_socket(&socket, key).is_ok() {
                 sockets.insert(key, socket);
                 assert_matches!(node.set_socket(key, socket), Ok(()));
                 assert_matches!(sanitize_entries(&node.addrs, &node.sockets), Ok(()));
@@ -877,7 +892,7 @@ mod tests {
                     assert_eq!(
                         node.cache[usize::from(key)],
                         socket
-                            .filter(|socket| sanitize_socket(socket).is_ok())
+                            .filter(|socket| sanitize_socket(socket, key).is_ok())
                             .copied()
                             .unwrap_or(SOCKET_ADDR_UNSPECIFIED),
                     );
@@ -1036,7 +1051,7 @@ mod tests {
     fn test_new_with_socketaddr() {
         let mut rng = rand::thread_rng();
         let socket = repeat_with(|| new_rand_socket(&mut rng))
-            .filter(|socket| matches!(sanitize_socket(socket), Ok(())))
+            .filter(|socket| matches!(sanitize_socket(socket, SOCKET_TAG_GOSSIP), Ok(())))
             .find(|socket| socket.port().checked_add(11).is_some())
             .unwrap();
         let node = ContactInfo::new_with_socketaddr(&Keypair::new().pubkey(), &socket);
@@ -1047,7 +1062,7 @@ mod tests {
     fn test_sanitize_quic_offset() {
         let mut rng = rand::thread_rng();
         let socket = repeat_with(|| new_rand_socket(&mut rng))
-            .filter(|socket| matches!(sanitize_socket(socket), Ok(())))
+            .filter(|socket| matches!(sanitize_socket(socket, SOCKET_TAG_GOSSIP), Ok(())))
             .find(|socket| socket.port().checked_add(QUIC_PORT_OFFSET).is_some())
             .unwrap();
         let mut other = get_quic_socket(&socket).unwrap();
@@ -1079,7 +1094,7 @@ mod tests {
             rng.gen(), // shred_version
         );
         let socket = repeat_with(|| new_rand_socket(&mut rng))
-            .filter(|socket| matches!(sanitize_socket(socket), Ok(())))
+            .filter(|socket| matches!(sanitize_socket(socket, SOCKET_TAG_GOSSIP), Ok(())))
             .find(|socket| socket.port().checked_add(QUIC_PORT_OFFSET).is_some())
             .unwrap();
         // TPU socket.

gossip/src/legacy_contact_info.rs

@@ -53,7 +53,7 @@ macro_rules! get_socket {
         #[cfg(test)]
         pub(crate) fn $name(&self) -> Option<SocketAddr> {
             let socket = self.$name;
-            sanitize_socket(&socket).ok()?;
+            sanitize_socket(&socket, 0).ok()?;
             Some(socket)
         }
     };
@@ -64,7 +64,7 @@ macro_rules! get_socket {
                 Protocol::QUIC => self.$quic,
                 Protocol::UDP => self.$name,
             };
-            sanitize_socket(&socket).ok()?;
+            sanitize_socket(&socket, 0).ok()?;
             Some(socket)
         }
     };
@@ -126,7 +126,7 @@ impl LegacyContactInfo {
 
     pub(crate) fn gossip(&self) -> Option<SocketAddr> {
         let socket = self.gossip;
-        crate::contact_info::sanitize_socket(&socket).ok()?;
+        crate::contact_info::sanitize_socket(&socket, 0).ok()?;
         Some(socket)
     }