Fix interface bind

gossip/src/main.rs

@@ -171,7 +171,7 @@ fn parse_gossip_host(matches: &ArgMatches, entrypoint_addr: Option<SocketAddr>)
         })
         .unwrap_or_else(|| {
             if let Some(entrypoint_addr) = entrypoint_addr {
-                solana_net_utils::get_public_ip_addr(&entrypoint_addr).unwrap_or_else(|err| {
+                solana_net_utils::get_public_ip_addr(&entrypoint_addr, None).unwrap_or_else(|err| {
                     eprintln!("Failed to contact cluster entrypoint {entrypoint_addr}: {err}");
                     exit(1);
                 })
@@ -222,7 +222,7 @@ fn get_entrypoint_shred_version(entrypoint: &Option<SocketAddr>) -> Option<u16>
         error!("cannot obtain shred-version without an entrypoint");
         return None;
     };
-    match solana_net_utils::get_cluster_shred_version(entrypoint) {
+    match solana_net_utils::get_cluster_shred_version(entrypoint, None) {
         Err(err) => {
             error!("get_cluster_shred_version failed: {entrypoint}, {err}");
             None

net-utils/Cargo.toml

@@ -10,7 +10,11 @@ license = { workspace = true }
 edition = { workspace = true }
 
 [dependencies]
+anyhow = { workspace = true }
+
 bincode = { workspace = true }
+bytes = { workspace = true }
+
 clap = { version = "3.1.5", features = ["cargo"], optional = true }
 crossbeam-channel = { workspace = true }
 log = { workspace = true }

net-utils/src/bin/ip_address.rs

@@ -16,7 +16,7 @@ fn main() {
     let addr = solana_net_utils::parse_host_port(host_port)
         .unwrap_or_else(|_| panic!("failed to parse {host_port}"));
 
-    match solana_net_utils::get_public_ip_addr(&addr) {
+    match solana_net_utils::get_public_ip_addr(&addr, None) {
         Ok(ip) => println!("{ip}"),
         Err(err) => {
             eprintln!("{addr}: {err}");

net-utils/src/lib.rs

@@ -1,19 +1,26 @@
 //! The `net_utils` module assists with networking
 #![allow(clippy::arithmetic_side_effects)]
+
 #[cfg(feature = "dev-context-only-utils")]
 use tokio::net::UdpSocket as TokioUdpSocket;
 use {
+    anyhow::{anyhow, bail},
+    bytes::{BufMut, BytesMut},
     crossbeam_channel::unbounded,
     log::*,
     rand::{thread_rng, Rng},
     socket2::{Domain, SockAddr, Socket, Type},
     std::{
         collections::{BTreeMap, HashSet},
-        io::{self, Read, Write},
+        io::{self},
         net::{IpAddr, Ipv4Addr, SocketAddr, TcpListener, TcpStream, ToSocketAddrs, UdpSocket},
         sync::{Arc, RwLock},
         time::{Duration, Instant},
     },
+    tokio::{
+        io::{AsyncReadExt, AsyncWriteExt},
+        net::TcpSocket,
+    },
     url::Url,
 };
 
@@ -39,90 +46,115 @@ pub const MINIMUM_VALIDATOR_PORT_RANGE_WIDTH: u16 = 17; // VALIDATOR_PORT_RANGE
 pub(crate) const HEADER_LENGTH: usize = 4;
 pub(crate) const IP_ECHO_SERVER_RESPONSE_LENGTH: usize = HEADER_LENGTH + 23;
 
-fn ip_echo_server_request(
-    ip_echo_server_addr: &SocketAddr,
+async fn ip_echo_server_request(
+    ip_echo_server_addr: SocketAddr,
     msg: IpEchoServerMessage,
-) -> Result<IpEchoServerResponse, String> {
+    bind_address: Option<IpAddr>,
+) -> anyhow::Result<IpEchoServerResponse> {
     let timeout = Duration::new(5, 0);
-    TcpStream::connect_timeout(ip_echo_server_addr, timeout)
-        .and_then(|mut stream| {
-            // Start with HEADER_LENGTH null bytes to avoid looking like an HTTP GET/POST request
-            let mut bytes = vec![0; HEADER_LENGTH];
-
-            bytes.append(&mut bincode::serialize(&msg).expect("serialize IpEchoServerMessage"));
-
-            // End with '\n' to make this request look HTTP-ish and tickle an error response back
-            // from an HTTP server
-            bytes.push(b'\n');
-
-            stream.set_read_timeout(Some(Duration::new(10, 0)))?;
-            stream.write_all(&bytes)?;
-            stream.shutdown(std::net::Shutdown::Write)?;
-            let mut data = vec![0u8; IP_ECHO_SERVER_RESPONSE_LENGTH];
-            let _ = stream.read(&mut data[..])?;
-            Ok(data)
-        })
-        .and_then(|data| {
-            // It's common for users to accidentally confuse the validator's gossip port and JSON
-            // RPC port.  Attempt to detect when this occurs by looking for the standard HTTP
-            // response header and provide the user with a helpful error message
-            if data.len() < HEADER_LENGTH {
-                return Err(io::Error::new(
-                    io::ErrorKind::Other,
-                    format!("Response too short, received {} bytes", data.len()),
-                ));
-            }
+    let socket = tokio::net::TcpSocket::new_v4()?;
+    if let Some(addr) = bind_address {
+        socket.bind(SocketAddr::new(addr, 0))?;
+    }
 
-            let response_header: String =
-                data[0..HEADER_LENGTH].iter().map(|b| *b as char).collect();
-            if response_header != "\0\0\0\0" {
-                if response_header == "HTTP" {
-                    let http_response = data.iter().map(|b| *b as char).collect::<String>();
-                    return Err(io::Error::new(
-                        io::ErrorKind::Other,
-                        format!(
-                            "Invalid gossip entrypoint. {ip_echo_server_addr} looks to be an HTTP port: {http_response}"
-                        ),
-                    ));
-                }
-                return Err(io::Error::new(
-                    io::ErrorKind::Other,
-                    format!(
-                        "Invalid gossip entrypoint. {ip_echo_server_addr} provided an invalid response header: '{response_header}'"
-                    ),
-                ));
+    async fn do_make_request(
+        socket: TcpSocket,
+        ip_echo_server_addr: SocketAddr,
+        msg: IpEchoServerMessage,
+    ) -> anyhow::Result<BytesMut> {
+        let mut stream = socket.connect(ip_echo_server_addr).await?;
+        // Start with HEADER_LENGTH null bytes to avoid looking like an HTTP GET/POST request
+        let mut bytes = BytesMut::with_capacity(IP_ECHO_SERVER_RESPONSE_LENGTH);
+        bytes.extend_from_slice(&[0u8; HEADER_LENGTH]);
+        bytes.extend_from_slice(&bincode::serialize(&msg)?);
+
+        // End with '\n' to make this request look HTTP-ish and tickle an error response back
+        // from an HTTP server
+        bytes.put_u8(b'\n');
+        stream.write_all(&bytes).await?;
+        stream.flush().await?;
+
+        bytes.clear();
+        let _n = stream.read_buf(&mut bytes).await?;
+        stream.shutdown().await?;
+
+        Ok(bytes)
+    }
+
+    let response =
+        tokio::time::timeout(timeout, do_make_request(socket, ip_echo_server_addr, msg)).await??;
+    // It's common for users to accidentally confuse the validator's gossip port and JSON
+    // RPC port.  Attempt to detect when this occurs by looking for the standard HTTP
+    // response header and provide the user with a helpful error message
+    if response.len() < HEADER_LENGTH {
+        bail!("Response too short, received {} bytes", response.len());
+    }
+
+    let (response_header, body) =
+        response
+            .split_first_chunk::<HEADER_LENGTH>()
+            .ok_or(anyhow::anyhow!(
+                "Not enough data in the response from {ip_echo_server_addr}!"
+            ))?;
+    let payload = match response_header {
+        [0, 0, 0, 0] => bincode::deserialize(&response[HEADER_LENGTH..])?,
+        [b'H', b'T', b'T', b'P'] => {
+            let http_response = std::str::from_utf8(body);
+            match http_response {
+                Ok(r) => bail!("Invalid gossip entrypoint. {ip_echo_server_addr} looks to be an HTTP port replying with {r}"),
+                Err(_) => bail!("Invalid gossip entrypoint. {ip_echo_server_addr} looks to be an HTTP port."),
             }
+        }
+        _ => {
+            bail!("Invalid gossip entrypoint. {ip_echo_server_addr} provided unexpected header bytes {response_header:?} ");
+        }
+    };
 
-            bincode::deserialize(&data[HEADER_LENGTH..]).map_err(|err| {
-                io::Error::new(
-                    io::ErrorKind::Other,
-                    format!("Failed to deserialize: {err:?}"),
-                )
-            })
-        })
-        .map_err(|err| err.to_string())
+    Ok(payload)
 }
 
 /// Determine the public IP address of this machine by asking an ip_echo_server at the given
 /// address
-pub fn get_public_ip_addr(ip_echo_server_addr: &SocketAddr) -> Result<IpAddr, String> {
-    let resp = ip_echo_server_request(ip_echo_server_addr, IpEchoServerMessage::default())?;
+pub fn get_public_ip_addr(
+    ip_echo_server_addr: &SocketAddr,
+    bind_address: Option<IpAddr>,
+) -> anyhow::Result<IpAddr> {
+    let fut = ip_echo_server_request(
+        *ip_echo_server_addr,
+        IpEchoServerMessage::default(),
+        bind_address,
+    );
+    let rt = tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .build()?;
+    let resp = rt.block_on(fut)?;
     Ok(resp.address)
 }
 
-pub fn get_cluster_shred_version(ip_echo_server_addr: &SocketAddr) -> Result<u16, String> {
-    let resp = ip_echo_server_request(ip_echo_server_addr, IpEchoServerMessage::default())?;
+pub fn get_cluster_shred_version(
+    ip_echo_server_addr: &SocketAddr,
+    bind_address: Option<IpAddr>,
+) -> anyhow::Result<u16> {
+    let fut = ip_echo_server_request(
+        *ip_echo_server_addr,
+        IpEchoServerMessage::default(),
+        bind_address,
+    );
+    let rt = tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .build()?;
+    let resp = rt.block_on(fut)?;
     resp.shred_version
-        .ok_or_else(|| String::from("IP echo server does not return a shred-version"))
+        .ok_or_else(|| anyhow!("IP echo server does not return a shred-version"))
 }
 
 // Checks if any of the provided TCP/UDP ports are not reachable by the machine at
 // `ip_echo_server_addr`
 const DEFAULT_TIMEOUT_SECS: u64 = 5;
 const DEFAULT_RETRY_COUNT: usize = 5;
 
-fn do_verify_reachable_ports(
-    ip_echo_server_addr: &SocketAddr,
+async fn do_verify_reachable_ports(
+    ip_echo_server_addr: SocketAddr,
     tcp_listeners: Vec<(u16, TcpListener)>,
     udp_sockets: &[&UdpSocket],
     timeout: u64,
@@ -137,7 +169,9 @@ fn do_verify_reachable_ports(
     let _ = ip_echo_server_request(
         ip_echo_server_addr,
         IpEchoServerMessage::new(&tcp_ports, &[]),
+        None,
     )
+    .await
     .map_err(|err| warn!("ip_echo_server request failed: {}", err));
 
     let mut ok = true;
@@ -218,7 +252,9 @@ fn do_verify_reachable_ports(
             let _ = ip_echo_server_request(
                 ip_echo_server_addr,
                 IpEchoServerMessage::new(&[], &checked_ports),
+                None,
             )
+            .await
             .map_err(|err| warn!("ip_echo_server request failed: {}", err));
 
             // Spawn threads at once!
@@ -300,13 +336,18 @@ pub fn verify_reachable_ports(
     tcp_listeners: Vec<(u16, TcpListener)>,
     udp_sockets: &[&UdpSocket],
 ) -> bool {
-    do_verify_reachable_ports(
-        ip_echo_server_addr,
+    let rt = tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .build()
+        .expect("Can not create a runtime");
+    let fut = do_verify_reachable_ports(
+        *ip_echo_server_addr,
         tcp_listeners,
         udp_sockets,
         DEFAULT_TIMEOUT_SECS,
         DEFAULT_RETRY_COUNT,
-    )
+    );
+    rt.block_on(fut)
 }
 
 pub fn parse_port_or_addr(optstr: Option<&str>, default_addr: SocketAddr) -> SocketAddr {
@@ -780,8 +821,14 @@ pub fn bind_more_with_config(
 
 #[cfg(test)]
 mod tests {
-    use {super::*, std::net::Ipv4Addr};
+    use {super::*, std::net::Ipv4Addr, tokio::runtime::Runtime};
 
+    fn runtime() -> Runtime {
+        tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("Can not create a runtime")
+    }
     #[test]
     fn test_response_length() {
         let resp = IpEchoServerResponse {
@@ -957,10 +1004,13 @@ mod tests {
 
         let server_ip_echo_addr = server_udp_socket.local_addr().unwrap();
         assert_eq!(
-            get_public_ip_addr(&server_ip_echo_addr),
-            parse_host("127.0.0.1"),
+            get_public_ip_addr(&server_ip_echo_addr, None).unwrap(),
+            parse_host("127.0.0.1").unwrap(),
+        );
+        assert_eq!(
+            get_cluster_shred_version(&server_ip_echo_addr, None).unwrap(),
+            42
         );
-        assert_eq!(get_cluster_shred_version(&server_ip_echo_addr), Ok(42));
         assert!(verify_reachable_ports(&server_ip_echo_addr, vec![], &[],));
     }
 
@@ -982,10 +1032,13 @@ mod tests {
 
         let ip_echo_server_addr = server_udp_socket.local_addr().unwrap();
         assert_eq!(
-            get_public_ip_addr(&ip_echo_server_addr),
-            parse_host("127.0.0.1"),
+            get_public_ip_addr(&ip_echo_server_addr, None).unwrap(),
+            parse_host("127.0.0.1").unwrap(),
+        );
+        assert_eq!(
+            get_cluster_shred_version(&ip_echo_server_addr, None).unwrap(),
+            65535
         );
-        assert_eq!(get_cluster_shred_version(&ip_echo_server_addr), Ok(65535));
         assert!(verify_reachable_ports(
             &ip_echo_server_addr,
             vec![(client_port, client_tcp_listener)],
@@ -1008,13 +1061,14 @@ mod tests {
         let (correct_client_port, (_client_udp_socket, client_tcp_listener)) =
             bind_common_in_range_with_config(ip_addr, (3200, 3250), config).unwrap();
 
-        assert!(!do_verify_reachable_ports(
-            &server_ip_echo_addr,
+        let rt = runtime();
+        assert!(!rt.block_on(do_verify_reachable_ports(
+            server_ip_echo_addr,
             vec![(correct_client_port, client_tcp_listener)],
             &[],
             2,
             3,
-        ));
+        )));
     }
 
     #[test]
@@ -1032,13 +1086,14 @@ mod tests {
         let (_correct_client_port, (client_udp_socket, _client_tcp_listener)) =
             bind_common_in_range_with_config(ip_addr, (3200, 3250), config).unwrap();
 
-        assert!(!do_verify_reachable_ports(
-            &server_ip_echo_addr,
+        let rt = runtime();
+        assert!(!rt.block_on(do_verify_reachable_ports(
+            server_ip_echo_addr,
             vec![],
             &[&client_udp_socket],
             2,
             3,
-        ));
+        )));
     }
 
     #[test]