adds validator flag to allow private ip addresses (#18850)

streamer/src/packet.rs

@@ -1,7 +1,7 @@
 //! The `packet` module defines data structures and methods to pull data from the network.
 use crate::{
     recvmmsg::{recv_mmsg, NUM_RCVMMSGS},
-    socket::is_global,
+    socket::SocketAddrSpace,
 };
 pub use solana_perf::packet::{
     limited_deserialize, to_packets_chunked, Packets, PacketsRecycler, NUM_PACKETS,
@@ -57,10 +57,14 @@ pub fn recv_from(obj: &mut Packets, socket: &UdpSocket, max_wait_ms: u64) -> Res
     Ok(i)
 }
 
-pub fn send_to(obj: &Packets, socket: &UdpSocket) -> Result<()> {
+pub fn send_to(
+    obj: &Packets,
+    socket: &UdpSocket,
+    socket_addr_space: &SocketAddrSpace,
+) -> Result<()> {
     for p in &obj.packets {
         let addr = p.meta.addr();
-        if is_global(&addr) {
+        if socket_addr_space.check(&addr) {
             socket.send_to(&p.data[..p.meta.size], &addr)?;
         }
     }
@@ -99,7 +103,7 @@ mod tests {
             m.meta.set_addr(&addr);
             m.meta.size = PACKET_DATA_SIZE;
         }
-        send_to(&p, &send_socket).unwrap();
+        send_to(&p, &send_socket, &SocketAddrSpace::Unspecified).unwrap();
 
         let recvd = recv_from(&mut p, &recv_socket, 1).unwrap();
 
@@ -152,7 +156,7 @@ mod tests {
                 m.meta.set_addr(&addr);
                 m.meta.size = 1;
             }
-            send_to(&p, &send_socket).unwrap();
+            send_to(&p, &send_socket, &SocketAddrSpace::Unspecified).unwrap();
         }
 
         let recvd = recv_from(&mut p, &recv_socket, 100).unwrap();

streamer/src/socket.rs

@@ -1,28 +1,39 @@
-use std::net::SocketAddr;
+use std::net::{IpAddr, SocketAddr};
 
-// TODO: remove these once IpAddr::is_global is stable.
-
-#[cfg(test)]
-pub fn is_global(_: &SocketAddr) -> bool {
-    true
+#[derive(Clone, Copy, PartialEq)]
+pub enum SocketAddrSpace {
+    Unspecified,
+    Global,
 }
 
-#[cfg(not(test))]
-pub fn is_global(addr: &SocketAddr) -> bool {
-    use std::net::IpAddr;
-
-    match addr.ip() {
-        IpAddr::V4(addr) => {
-            // TODO: Consider excluding:
-            //    addr.is_loopback() || addr.is_link_local()
-            // || addr.is_broadcast() || addr.is_documentation()
-            // || addr.is_unspecified()
-            !addr.is_private()
+impl SocketAddrSpace {
+    pub fn new(allow_private_addr: bool) -> Self {
+        if allow_private_addr {
+            SocketAddrSpace::Unspecified
+        } else {
+            SocketAddrSpace::Global
         }
-        IpAddr::V6(_) => {
-            // TODO: Consider excluding:
-            // addr.is_loopback() || addr.is_unspecified(),
-            true
+    }
+
+    /// Returns true if the IP address is valid.
+    pub fn check(&self, addr: &SocketAddr) -> bool {
+        if self == &SocketAddrSpace::Unspecified {
+            return true;
+        }
+        // TODO: remove these once IpAddr::is_global is stable.
+        match addr.ip() {
+            IpAddr::V4(addr) => {
+                // TODO: Consider excluding:
+                //    addr.is_loopback() || addr.is_link_local()
+                // || addr.is_broadcast() || addr.is_documentation()
+                // || addr.is_unspecified()
+                !addr.is_private()
+            }
+            IpAddr::V6(_) => {
+                // TODO: Consider excluding:
+                // addr.is_loopback() || addr.is_unspecified(),
+                true
+            }
         }
     }
 }

streamer/src/streamer.rs

@@ -1,8 +1,11 @@
 //! The `streamer` module defines a set of services for efficiently pulling data from UDP sockets.
 //!
 
-use crate::packet::{self, send_to, Packets, PacketsRecycler, PACKETS_PER_BATCH};
-use crate::recvmmsg::NUM_RCVMMSGS;
+use crate::{
+    packet::{self, send_to, Packets, PacketsRecycler, PACKETS_PER_BATCH},
+    recvmmsg::NUM_RCVMMSGS,
+    socket::SocketAddrSpace,
+};
 use solana_sdk::timing::{duration_as_ms, timestamp};
 use std::net::UdpSocket;
 use std::sync::atomic::{AtomicBool, Ordering};
@@ -112,10 +115,14 @@ pub fn receiver(
         .unwrap()
 }
 
-fn recv_send(sock: &UdpSocket, r: &PacketReceiver) -> Result<()> {
+fn recv_send(
+    sock: &UdpSocket,
+    r: &PacketReceiver,
+    socket_addr_space: &SocketAddrSpace,
+) -> Result<()> {
     let timer = Duration::new(1, 0);
     let msgs = r.recv_timeout(timer)?;
-    send_to(&msgs, sock)?;
+    send_to(&msgs, sock, socket_addr_space)?;
     Ok(())
 }
 
@@ -138,7 +145,12 @@ pub fn recv_batch(recvr: &PacketReceiver, max_batch: usize) -> Result<(Vec<Packe
     Ok((batch, len, duration_as_ms(&recv_start.elapsed())))
 }
 
-pub fn responder(name: &'static str, sock: Arc<UdpSocket>, r: PacketReceiver) -> JoinHandle<()> {
+pub fn responder(
+    name: &'static str,
+    sock: Arc<UdpSocket>,
+    r: PacketReceiver,
+    socket_addr_space: SocketAddrSpace,
+) -> JoinHandle<()> {
     Builder::new()
         .name(format!("solana-responder-{}", name))
         .spawn(move || {
@@ -146,7 +158,7 @@ pub fn responder(name: &'static str, sock: Arc<UdpSocket>, r: PacketReceiver) ->
             let mut last_error = None;
             let mut last_print = 0;
             loop {
-                if let Err(e) = recv_send(&sock, &r) {
+                if let Err(e) = recv_send(&sock, &r, &socket_addr_space) {
                     match e {
                         StreamerError::RecvTimeout(RecvTimeoutError::Disconnected) => break,
                         StreamerError::RecvTimeout(RecvTimeoutError::Timeout) => (),
@@ -222,7 +234,12 @@ mod test {
         );
         let t_responder = {
             let (s_responder, r_responder) = channel();
-            let t_responder = responder("streamer_send_test", Arc::new(send), r_responder);
+            let t_responder = responder(
+                "streamer_send_test",
+                Arc::new(send),
+                r_responder,
+                SocketAddrSpace::Unspecified,
+            );
             let mut msgs = Packets::default();
             for i in 0..5 {
                 let mut b = Packet::default();