solana_net_utils/
lib.rs

1//! The `net_utils` module assists with networking
2#![allow(clippy::arithmetic_side_effects)]
3use {
4    crossbeam_channel::unbounded,
5    log::*,
6    rand::{thread_rng, Rng},
7    socket2::{Domain, SockAddr, Socket, Type},
8    std::{
9        collections::{BTreeMap, HashSet},
10        io::{self, Read, Write},
11        net::{IpAddr, SocketAddr, TcpListener, TcpStream, ToSocketAddrs, UdpSocket},
12        sync::{Arc, RwLock},
13        time::{Duration, Instant},
14    },
15    url::Url,
16};
17
18mod ip_echo_server;
19pub use ip_echo_server::{
20    ip_echo_server, IpEchoServer, DEFAULT_IP_ECHO_SERVER_THREADS, MAX_PORT_COUNT_PER_MESSAGE,
21    MINIMUM_IP_ECHO_SERVER_THREADS,
22};
23use ip_echo_server::{IpEchoServerMessage, IpEchoServerResponse};
24
25/// A data type representing a public Udp socket
26pub struct UdpSocketPair {
27    pub addr: SocketAddr,    // Public address of the socket
28    pub receiver: UdpSocket, // Locally bound socket that can receive from the public address
29    pub sender: UdpSocket,   // Locally bound socket to send via public address
30}
31
32pub type PortRange = (u16, u16);
33
34pub const VALIDATOR_PORT_RANGE: PortRange = (8000, 10_000);
35pub const MINIMUM_VALIDATOR_PORT_RANGE_WIDTH: u16 = 16; // VALIDATOR_PORT_RANGE must be at least this wide
36
37pub(crate) const HEADER_LENGTH: usize = 4;
38pub(crate) const IP_ECHO_SERVER_RESPONSE_LENGTH: usize = HEADER_LENGTH + 23;
39
40fn ip_echo_server_request(
41    ip_echo_server_addr: &SocketAddr,
42    msg: IpEchoServerMessage,
43) -> Result<IpEchoServerResponse, String> {
44    let timeout = Duration::new(5, 0);
45    TcpStream::connect_timeout(ip_echo_server_addr, timeout)
46        .and_then(|mut stream| {
47            // Start with HEADER_LENGTH null bytes to avoid looking like an HTTP GET/POST request
48            let mut bytes = vec![0; HEADER_LENGTH];
49
50            bytes.append(&mut bincode::serialize(&msg).expect("serialize IpEchoServerMessage"));
51
52            // End with '\n' to make this request look HTTP-ish and tickle an error response back
53            // from an HTTP server
54            bytes.push(b'\n');
55
56            stream.set_read_timeout(Some(Duration::new(10, 0)))?;
57            stream.write_all(&bytes)?;
58            stream.shutdown(std::net::Shutdown::Write)?;
59            let mut data = vec![0u8; IP_ECHO_SERVER_RESPONSE_LENGTH];
60            let _ = stream.read(&mut data[..])?;
61            Ok(data)
62        })
63        .and_then(|data| {
64            // It's common for users to accidentally confuse the validator's gossip port and JSON
65            // RPC port.  Attempt to detect when this occurs by looking for the standard HTTP
66            // response header and provide the user with a helpful error message
67            if data.len() < HEADER_LENGTH {
68                return Err(io::Error::new(
69                    io::ErrorKind::Other,
70                    format!("Response too short, received {} bytes", data.len()),
71                ));
72            }
73
74            let response_header: String =
75                data[0..HEADER_LENGTH].iter().map(|b| *b as char).collect();
76            if response_header != "\0\0\0\0" {
77                if response_header == "HTTP" {
78                    let http_response = data.iter().map(|b| *b as char).collect::<String>();
79                    return Err(io::Error::new(
80                        io::ErrorKind::Other,
81                        format!(
82                            "Invalid gossip entrypoint. {ip_echo_server_addr} looks to be an HTTP port: {http_response}"
83                        ),
84                    ));
85                }
86                return Err(io::Error::new(
87                    io::ErrorKind::Other,
88                    format!(
89                        "Invalid gossip entrypoint. {ip_echo_server_addr} provided an invalid response header: '{response_header}'"
90                    ),
91                ));
92            }
93
94            bincode::deserialize(&data[HEADER_LENGTH..]).map_err(|err| {
95                io::Error::new(
96                    io::ErrorKind::Other,
97                    format!("Failed to deserialize: {err:?}"),
98                )
99            })
100        })
101        .map_err(|err| err.to_string())
102}
103
104/// Determine the public IP address of this machine by asking an ip_echo_server at the given
105/// address
106pub fn get_public_ip_addr(ip_echo_server_addr: &SocketAddr) -> Result<IpAddr, String> {
107    let resp = ip_echo_server_request(ip_echo_server_addr, IpEchoServerMessage::default())?;
108    Ok(resp.address)
109}
110
111pub fn get_cluster_shred_version(ip_echo_server_addr: &SocketAddr) -> Result<u16, String> {
112    let resp = ip_echo_server_request(ip_echo_server_addr, IpEchoServerMessage::default())?;
113    resp.shred_version
114        .ok_or_else(|| String::from("IP echo server does not return a shred-version"))
115}
116
117// Checks if any of the provided TCP/UDP ports are not reachable by the machine at
118// `ip_echo_server_addr`
119const DEFAULT_TIMEOUT_SECS: u64 = 5;
120const DEFAULT_RETRY_COUNT: usize = 5;
121
122fn do_verify_reachable_ports(
123    ip_echo_server_addr: &SocketAddr,
124    tcp_listeners: Vec<(u16, TcpListener)>,
125    udp_sockets: &[&UdpSocket],
126    timeout: u64,
127    udp_retry_count: usize,
128) -> bool {
129    info!(
130        "Checking that tcp ports {:?} are reachable from {:?}",
131        tcp_listeners, ip_echo_server_addr
132    );
133
134    let tcp_ports: Vec<_> = tcp_listeners.iter().map(|(port, _)| *port).collect();
135    let _ = ip_echo_server_request(
136        ip_echo_server_addr,
137        IpEchoServerMessage::new(&tcp_ports, &[]),
138    )
139    .map_err(|err| warn!("ip_echo_server request failed: {}", err));
140
141    let mut ok = true;
142    let timeout = Duration::from_secs(timeout);
143
144    // Wait for a connection to open on each TCP port
145    for (port, tcp_listener) in tcp_listeners {
146        let (sender, receiver) = unbounded();
147        let listening_addr = tcp_listener.local_addr().unwrap();
148        let thread_handle = std::thread::Builder::new()
149            .name(format!("solVrfyTcp{port:05}"))
150            .spawn(move || {
151                debug!("Waiting for incoming connection on tcp/{}", port);
152                match tcp_listener.incoming().next() {
153                    Some(_) => sender
154                        .send(())
155                        .unwrap_or_else(|err| warn!("send failure: {}", err)),
156                    None => warn!("tcp incoming failed"),
157                }
158            })
159            .unwrap();
160        match receiver.recv_timeout(timeout) {
161            Ok(_) => {
162                info!("tcp/{} is reachable", port);
163            }
164            Err(err) => {
165                error!(
166                    "Received no response at tcp/{}, check your port configuration: {}",
167                    port, err
168                );
169                // Ugh, std rustc doesn't provide accepting with timeout or restoring original
170                // nonblocking-status of sockets because of lack of getter, only the setter...
171                // So, to close the thread cleanly, just connect from here.
172                // ref: https://github.com/rust-lang/rust/issues/31615
173                TcpStream::connect_timeout(&listening_addr, timeout).unwrap();
174                ok = false;
175            }
176        }
177        // ensure to reap the thread
178        thread_handle.join().unwrap();
179    }
180
181    if !ok {
182        // No retries for TCP, abort on the first failure
183        return ok;
184    }
185
186    let mut udp_ports: BTreeMap<_, _> = BTreeMap::new();
187    udp_sockets.iter().for_each(|udp_socket| {
188        let port = udp_socket.local_addr().unwrap().port();
189        udp_ports
190            .entry(port)
191            .or_insert_with(Vec::new)
192            .push(udp_socket);
193    });
194    let udp_ports: Vec<_> = udp_ports.into_iter().collect();
195
196    info!(
197        "Checking that udp ports {:?} are reachable from {:?}",
198        udp_ports.iter().map(|(port, _)| port).collect::<Vec<_>>(),
199        ip_echo_server_addr
200    );
201
202    'outer: for checked_ports_and_sockets in udp_ports.chunks(MAX_PORT_COUNT_PER_MESSAGE) {
203        ok = false;
204
205        for udp_remaining_retry in (0_usize..udp_retry_count).rev() {
206            let (checked_ports, checked_socket_iter) = (
207                checked_ports_and_sockets
208                    .iter()
209                    .map(|(port, _)| *port)
210                    .collect::<Vec<_>>(),
211                checked_ports_and_sockets
212                    .iter()
213                    .flat_map(|(_, sockets)| sockets),
214            );
215
216            let _ = ip_echo_server_request(
217                ip_echo_server_addr,
218                IpEchoServerMessage::new(&[], &checked_ports),
219            )
220            .map_err(|err| warn!("ip_echo_server request failed: {}", err));
221
222            // Spawn threads at once!
223            let reachable_ports = Arc::new(RwLock::new(HashSet::new()));
224            let thread_handles: Vec<_> = checked_socket_iter
225                .map(|udp_socket| {
226                    let port = udp_socket.local_addr().unwrap().port();
227                    let udp_socket = udp_socket.try_clone().expect("Unable to clone udp socket");
228                    let reachable_ports = reachable_ports.clone();
229
230                    std::thread::Builder::new()
231                        .name(format!("solVrfyUdp{port:05}"))
232                        .spawn(move || {
233                            let start = Instant::now();
234
235                            let original_read_timeout = udp_socket.read_timeout().unwrap();
236                            udp_socket
237                                .set_read_timeout(Some(Duration::from_millis(250)))
238                                .unwrap();
239                            loop {
240                                if reachable_ports.read().unwrap().contains(&port)
241                                    || Instant::now().duration_since(start) >= timeout
242                                {
243                                    break;
244                                }
245
246                                let recv_result = udp_socket.recv(&mut [0; 1]);
247                                debug!(
248                                    "Waited for incoming datagram on udp/{}: {:?}",
249                                    port, recv_result
250                                );
251
252                                if recv_result.is_ok() {
253                                    reachable_ports.write().unwrap().insert(port);
254                                    break;
255                                }
256                            }
257                            udp_socket.set_read_timeout(original_read_timeout).unwrap();
258                        })
259                        .unwrap()
260                })
261                .collect();
262
263            // Now join threads!
264            // Separate from the above by collect()-ing as an intermediately step to make the iterator
265            // eager not lazy so that joining happens here at once after creating bunch of threads
266            // at once.
267            for thread in thread_handles {
268                thread.join().unwrap();
269            }
270
271            let reachable_ports = reachable_ports.read().unwrap().clone();
272            if reachable_ports.len() == checked_ports.len() {
273                info!(
274                    "checked udp ports: {:?}, reachable udp ports: {:?}",
275                    checked_ports, reachable_ports
276                );
277                ok = true;
278                break;
279            } else if udp_remaining_retry > 0 {
280                // Might have lost a UDP packet, retry a couple times
281                error!(
282                    "checked udp ports: {:?}, reachable udp ports: {:?}",
283                    checked_ports, reachable_ports
284                );
285                error!("There are some udp ports with no response!! Retrying...");
286            } else {
287                error!("Maximum retry count is reached....");
288                break 'outer;
289            }
290        }
291    }
292
293    ok
294}
295
296pub fn verify_reachable_ports(
297    ip_echo_server_addr: &SocketAddr,
298    tcp_listeners: Vec<(u16, TcpListener)>,
299    udp_sockets: &[&UdpSocket],
300) -> bool {
301    do_verify_reachable_ports(
302        ip_echo_server_addr,
303        tcp_listeners,
304        udp_sockets,
305        DEFAULT_TIMEOUT_SECS,
306        DEFAULT_RETRY_COUNT,
307    )
308}
309
310pub fn parse_port_or_addr(optstr: Option<&str>, default_addr: SocketAddr) -> SocketAddr {
311    if let Some(addrstr) = optstr {
312        if let Ok(port) = addrstr.parse() {
313            let mut addr = default_addr;
314            addr.set_port(port);
315            addr
316        } else if let Ok(addr) = addrstr.parse() {
317            addr
318        } else {
319            default_addr
320        }
321    } else {
322        default_addr
323    }
324}
325
326pub fn parse_port_range(port_range: &str) -> Option<PortRange> {
327    let ports: Vec<&str> = port_range.split('-').collect();
328    if ports.len() != 2 {
329        return None;
330    }
331
332    let start_port = ports[0].parse();
333    let end_port = ports[1].parse();
334
335    if start_port.is_err() || end_port.is_err() {
336        return None;
337    }
338    let start_port = start_port.unwrap();
339    let end_port = end_port.unwrap();
340    if end_port < start_port {
341        return None;
342    }
343    Some((start_port, end_port))
344}
345
346pub fn parse_host(host: &str) -> Result<IpAddr, String> {
347    // First, check if the host syntax is valid. This check is needed because addresses
348    // such as `("localhost:1234", 0)` will resolve to IPs on some networks.
349    let parsed_url = Url::parse(&format!("http://{host}")).map_err(|e| e.to_string())?;
350    if parsed_url.port().is_some() {
351        return Err(format!("Expected port in URL: {host}"));
352    }
353
354    // Next, check to see if it resolves to an IP address
355    let ips: Vec<_> = (host, 0)
356        .to_socket_addrs()
357        .map_err(|err| err.to_string())?
358        .map(|socket_address| socket_address.ip())
359        .collect();
360    if ips.is_empty() {
361        Err(format!("Unable to resolve host: {host}"))
362    } else {
363        Ok(ips[0])
364    }
365}
366
367pub fn is_host(string: String) -> Result<(), String> {
368    parse_host(&string).map(|_| ())
369}
370
371pub fn parse_host_port(host_port: &str) -> Result<SocketAddr, String> {
372    let addrs: Vec<_> = host_port
373        .to_socket_addrs()
374        .map_err(|err| format!("Unable to resolve host {host_port}: {err}"))?
375        .collect();
376    if addrs.is_empty() {
377        Err(format!("Unable to resolve host: {host_port}"))
378    } else {
379        Ok(addrs[0])
380    }
381}
382
383pub fn is_host_port(string: String) -> Result<(), String> {
384    parse_host_port(&string).map(|_| ())
385}
386
387#[derive(Clone, Debug)]
388pub struct SocketConfig {
389    pub reuseport: bool,
390}
391
392impl Default for SocketConfig {
393    #[allow(clippy::derivable_impls)]
394    fn default() -> Self {
395        Self { reuseport: false }
396    }
397}
398
399#[cfg(any(windows, target_os = "ios"))]
400fn udp_socket(_reuseaddr: bool) -> io::Result<Socket> {
401    let sock = Socket::new(Domain::IPV4, Type::DGRAM, None)?;
402    Ok(sock)
403}
404
405#[cfg(any(windows, target_os = "ios"))]
406fn udp_socket_with_config(_config: SocketConfig) -> io::Result<Socket> {
407    let sock = Socket::new(Domain::IPV4, Type::DGRAM, None)?;
408    Ok(sock)
409}
410
411#[cfg(not(any(windows, target_os = "ios")))]
412fn udp_socket(reuseport: bool) -> io::Result<Socket> {
413    let config = SocketConfig { reuseport };
414    udp_socket_with_config(config)
415}
416
417#[cfg(not(any(windows, target_os = "ios")))]
418fn udp_socket_with_config(config: SocketConfig) -> io::Result<Socket> {
419    use nix::sys::socket::{setsockopt, sockopt::ReusePort};
420    let SocketConfig { reuseport } = config;
421
422    let sock = Socket::new(Domain::IPV4, Type::DGRAM, None)?;
423
424    if reuseport {
425        setsockopt(&sock, ReusePort, &true).ok();
426    }
427
428    Ok(sock)
429}
430
431// Find a port in the given range that is available for both TCP and UDP
432pub fn bind_common_in_range(
433    ip_addr: IpAddr,
434    range: PortRange,
435) -> io::Result<(u16, (UdpSocket, TcpListener))> {
436    for port in range.0..range.1 {
437        if let Ok((sock, listener)) = bind_common(ip_addr, port) {
438            return Result::Ok((sock.local_addr().unwrap().port(), (sock, listener)));
439        }
440    }
441
442    Err(io::Error::new(
443        io::ErrorKind::Other,
444        format!("No available TCP/UDP ports in {range:?}"),
445    ))
446}
447
448pub fn bind_in_range(ip_addr: IpAddr, range: PortRange) -> io::Result<(u16, UdpSocket)> {
449    let config = SocketConfig::default();
450    bind_in_range_with_config(ip_addr, range, config)
451}
452
453pub fn bind_in_range_with_config(
454    ip_addr: IpAddr,
455    range: PortRange,
456    config: SocketConfig,
457) -> io::Result<(u16, UdpSocket)> {
458    let sock = udp_socket_with_config(config)?;
459
460    for port in range.0..range.1 {
461        let addr = SocketAddr::new(ip_addr, port);
462
463        if sock.bind(&SockAddr::from(addr)).is_ok() {
464            let sock: UdpSocket = sock.into();
465            return Result::Ok((sock.local_addr().unwrap().port(), sock));
466        }
467    }
468
469    Err(io::Error::new(
470        io::ErrorKind::Other,
471        format!("No available UDP ports in {range:?}"),
472    ))
473}
474
475pub fn bind_with_any_port(ip_addr: IpAddr) -> io::Result<UdpSocket> {
476    let sock = udp_socket(false)?;
477    let addr = SocketAddr::new(ip_addr, 0);
478    match sock.bind(&SockAddr::from(addr)) {
479        Ok(_) => Result::Ok(sock.into()),
480        Err(err) => Err(io::Error::new(
481            io::ErrorKind::Other,
482            format!("No available UDP port: {err}"),
483        )),
484    }
485}
486
487// binds many sockets to the same port in a range
488pub fn multi_bind_in_range(
489    ip_addr: IpAddr,
490    range: PortRange,
491    mut num: usize,
492) -> io::Result<(u16, Vec<UdpSocket>)> {
493    if cfg!(windows) && num != 1 {
494        // See https://github.com/solana-labs/solana/issues/4607
495        warn!(
496            "multi_bind_in_range() only supports 1 socket in windows ({} requested)",
497            num
498        );
499        num = 1;
500    }
501    let mut sockets = Vec::with_capacity(num);
502
503    const NUM_TRIES: usize = 100;
504    let mut port = 0;
505    let mut error = None;
506    for _ in 0..NUM_TRIES {
507        port = {
508            let (port, _) = bind_in_range(ip_addr, range)?;
509            port
510        }; // drop the probe, port should be available... briefly.
511
512        let config = SocketConfig { reuseport: true };
513        for _ in 0..num {
514            let sock = bind_to_with_config(ip_addr, port, config.clone());
515            if let Ok(sock) = sock {
516                sockets.push(sock);
517            } else {
518                error = Some(sock);
519                break;
520            }
521        }
522        if sockets.len() == num {
523            break;
524        } else {
525            sockets.clear();
526        }
527    }
528    if sockets.len() != num {
529        error.unwrap()?;
530    }
531    Ok((port, sockets))
532}
533
534pub fn bind_to(ip_addr: IpAddr, port: u16, reuseport: bool) -> io::Result<UdpSocket> {
535    let config = SocketConfig { reuseport };
536    bind_to_with_config(ip_addr, port, config)
537}
538
539pub fn bind_to_with_config(
540    ip_addr: IpAddr,
541    port: u16,
542    config: SocketConfig,
543) -> io::Result<UdpSocket> {
544    let sock = udp_socket_with_config(config)?;
545
546    let addr = SocketAddr::new(ip_addr, port);
547
548    sock.bind(&SockAddr::from(addr)).map(|_| sock.into())
549}
550
551// binds both a UdpSocket and a TcpListener
552pub fn bind_common(ip_addr: IpAddr, port: u16) -> io::Result<(UdpSocket, TcpListener)> {
553    let config = SocketConfig { reuseport: false };
554    bind_common_with_config(ip_addr, port, config)
555}
556
557// binds both a UdpSocket and a TcpListener
558pub fn bind_common_with_config(
559    ip_addr: IpAddr,
560    port: u16,
561    config: SocketConfig,
562) -> io::Result<(UdpSocket, TcpListener)> {
563    let sock = udp_socket_with_config(config)?;
564
565    let addr = SocketAddr::new(ip_addr, port);
566    let sock_addr = SockAddr::from(addr);
567    sock.bind(&sock_addr)
568        .and_then(|_| TcpListener::bind(addr).map(|listener| (sock.into(), listener)))
569}
570
571pub fn bind_two_in_range_with_offset(
572    ip_addr: IpAddr,
573    range: PortRange,
574    offset: u16,
575) -> io::Result<((u16, UdpSocket), (u16, UdpSocket))> {
576    let sock1_config = SocketConfig::default();
577    let sock2_config = SocketConfig::default();
578    bind_two_in_range_with_offset_and_config(ip_addr, range, offset, sock1_config, sock2_config)
579}
580
581pub fn bind_two_in_range_with_offset_and_config(
582    ip_addr: IpAddr,
583    range: PortRange,
584    offset: u16,
585    sock1_config: SocketConfig,
586    sock2_config: SocketConfig,
587) -> io::Result<((u16, UdpSocket), (u16, UdpSocket))> {
588    if range.1.saturating_sub(range.0) < offset {
589        return Err(io::Error::new(
590            io::ErrorKind::Other,
591            "range too small to find two ports with the correct offset".to_string(),
592        ));
593    }
594    for port in range.0..range.1 {
595        if let Ok(first_bind) = bind_to_with_config(ip_addr, port, sock1_config.clone()) {
596            if range.1.saturating_sub(port) >= offset {
597                if let Ok(second_bind) =
598                    bind_to_with_config(ip_addr, port + offset, sock2_config.clone())
599                {
600                    return Ok((
601                        (first_bind.local_addr().unwrap().port(), first_bind),
602                        (second_bind.local_addr().unwrap().port(), second_bind),
603                    ));
604                }
605            } else {
606                break;
607            }
608        }
609    }
610    Err(io::Error::new(
611        io::ErrorKind::Other,
612        "couldn't find two ports with the correct offset in range".to_string(),
613    ))
614}
615
616pub fn find_available_port_in_range(ip_addr: IpAddr, range: PortRange) -> io::Result<u16> {
617    let (start, end) = range;
618    let mut tries_left = end - start;
619    let mut rand_port = thread_rng().gen_range(start..end);
620    loop {
621        match bind_common(ip_addr, rand_port) {
622            Ok(_) => {
623                break Ok(rand_port);
624            }
625            Err(err) => {
626                if tries_left == 0 {
627                    return Err(err);
628                }
629            }
630        }
631        rand_port += 1;
632        if rand_port == end {
633            rand_port = start;
634        }
635        tries_left -= 1;
636    }
637}
638
639pub fn bind_more_with_config(
640    socket: UdpSocket,
641    num: usize,
642    config: SocketConfig,
643) -> io::Result<Vec<UdpSocket>> {
644    let addr = socket.local_addr().unwrap();
645    let ip = addr.ip();
646    let port = addr.port();
647    std::iter::once(Ok(socket))
648        .chain((1..num).map(|_| bind_to_with_config(ip, port, config.clone())))
649        .collect()
650}
651
652#[cfg(test)]
653mod tests {
654    use {super::*, std::net::Ipv4Addr};
655
656    #[test]
657    fn test_response_length() {
658        let resp = IpEchoServerResponse {
659            address: IpAddr::from([u16::MAX; 8]), // IPv6 variant
660            shred_version: Some(u16::MAX),
661        };
662        let resp_size = bincode::serialized_size(&resp).unwrap();
663        assert_eq!(
664            IP_ECHO_SERVER_RESPONSE_LENGTH,
665            HEADER_LENGTH + resp_size as usize
666        );
667    }
668
669    // Asserts that an old client can parse the response from a new server.
670    #[test]
671    fn test_backward_compat() {
672        let address = IpAddr::from([
673            525u16, 524u16, 523u16, 522u16, 521u16, 520u16, 519u16, 518u16,
674        ]);
675        let response = IpEchoServerResponse {
676            address,
677            shred_version: Some(42),
678        };
679        let mut data = vec![0u8; IP_ECHO_SERVER_RESPONSE_LENGTH];
680        bincode::serialize_into(&mut data[HEADER_LENGTH..], &response).unwrap();
681        data.truncate(HEADER_LENGTH + 20);
682        assert_eq!(
683            bincode::deserialize::<IpAddr>(&data[HEADER_LENGTH..]).unwrap(),
684            address
685        );
686    }
687
688    // Asserts that a new client can parse the response from an old server.
689    #[test]
690    fn test_forward_compat() {
691        let address = IpAddr::from([
692            525u16, 524u16, 523u16, 522u16, 521u16, 520u16, 519u16, 518u16,
693        ]);
694        let mut data = [0u8; IP_ECHO_SERVER_RESPONSE_LENGTH];
695        bincode::serialize_into(&mut data[HEADER_LENGTH..], &address).unwrap();
696        let response: Result<IpEchoServerResponse, _> =
697            bincode::deserialize(&data[HEADER_LENGTH..]);
698        assert_eq!(
699            response.unwrap(),
700            IpEchoServerResponse {
701                address,
702                shred_version: None,
703            }
704        );
705    }
706
707    #[test]
708    fn test_parse_port_or_addr() {
709        let p1 = parse_port_or_addr(Some("9000"), SocketAddr::from(([1, 2, 3, 4], 1)));
710        assert_eq!(p1.port(), 9000);
711        let p2 = parse_port_or_addr(Some("127.0.0.1:7000"), SocketAddr::from(([1, 2, 3, 4], 1)));
712        assert_eq!(p2.port(), 7000);
713        let p2 = parse_port_or_addr(Some("hi there"), SocketAddr::from(([1, 2, 3, 4], 1)));
714        assert_eq!(p2.port(), 1);
715        let p3 = parse_port_or_addr(None, SocketAddr::from(([1, 2, 3, 4], 1)));
716        assert_eq!(p3.port(), 1);
717    }
718
719    #[test]
720    fn test_parse_port_range() {
721        assert_eq!(parse_port_range("garbage"), None);
722        assert_eq!(parse_port_range("1-"), None);
723        assert_eq!(parse_port_range("1-2"), Some((1, 2)));
724        assert_eq!(parse_port_range("1-2-3"), None);
725        assert_eq!(parse_port_range("2-1"), None);
726    }
727
728    #[test]
729    fn test_parse_host() {
730        parse_host("localhost:1234").unwrap_err();
731        parse_host("localhost").unwrap();
732        parse_host("127.0.0.0:1234").unwrap_err();
733        parse_host("127.0.0.0").unwrap();
734    }
735
736    #[test]
737    fn test_parse_host_port() {
738        parse_host_port("localhost:1234").unwrap();
739        parse_host_port("localhost").unwrap_err();
740        parse_host_port("127.0.0.0:1234").unwrap();
741        parse_host_port("127.0.0.0").unwrap_err();
742    }
743
744    #[test]
745    fn test_is_host_port() {
746        assert!(is_host_port("localhost:1234".to_string()).is_ok());
747        assert!(is_host_port("localhost".to_string()).is_err());
748    }
749
750    #[test]
751    fn test_bind() {
752        let ip_addr = IpAddr::V4(Ipv4Addr::UNSPECIFIED);
753        assert_eq!(bind_in_range(ip_addr, (2000, 2001)).unwrap().0, 2000);
754        let ip_addr = IpAddr::V4(Ipv4Addr::UNSPECIFIED);
755        let config = SocketConfig { reuseport: true };
756        let x = bind_to_with_config(ip_addr, 2002, config.clone()).unwrap();
757        let y = bind_to_with_config(ip_addr, 2002, config).unwrap();
758        assert_eq!(
759            x.local_addr().unwrap().port(),
760            y.local_addr().unwrap().port()
761        );
762        bind_to(ip_addr, 2002, false).unwrap_err();
763        bind_in_range(ip_addr, (2002, 2003)).unwrap_err();
764
765        let (port, v) = multi_bind_in_range(ip_addr, (2010, 2110), 10).unwrap();
766        for sock in &v {
767            assert_eq!(port, sock.local_addr().unwrap().port());
768        }
769    }
770
771    #[test]
772    fn test_bind_with_any_port() {
773        let ip_addr = IpAddr::V4(Ipv4Addr::UNSPECIFIED);
774        let x = bind_with_any_port(ip_addr).unwrap();
775        let y = bind_with_any_port(ip_addr).unwrap();
776        assert_ne!(
777            x.local_addr().unwrap().port(),
778            y.local_addr().unwrap().port()
779        );
780    }
781
782    #[test]
783    fn test_bind_in_range_nil() {
784        let ip_addr = IpAddr::V4(Ipv4Addr::UNSPECIFIED);
785        bind_in_range(ip_addr, (2000, 2000)).unwrap_err();
786        bind_in_range(ip_addr, (2000, 1999)).unwrap_err();
787    }
788
789    #[test]
790    fn test_find_available_port_in_range() {
791        let ip_addr = IpAddr::V4(Ipv4Addr::UNSPECIFIED);
792        assert_eq!(
793            find_available_port_in_range(ip_addr, (3000, 3001)).unwrap(),
794            3000
795        );
796        let port = find_available_port_in_range(ip_addr, (3000, 3050)).unwrap();
797        assert!((3000..3050).contains(&port));
798
799        let _socket = bind_to(ip_addr, port, false).unwrap();
800        find_available_port_in_range(ip_addr, (port, port + 1)).unwrap_err();
801    }
802
803    #[test]
804    fn test_bind_common_in_range() {
805        let ip_addr = IpAddr::V4(Ipv4Addr::UNSPECIFIED);
806        let (port, _sockets) = bind_common_in_range(ip_addr, (3100, 3150)).unwrap();
807        assert!((3100..3150).contains(&port));
808
809        bind_common_in_range(ip_addr, (port, port + 1)).unwrap_err();
810    }
811
812    #[test]
813    fn test_get_public_ip_addr_none() {
814        solana_logger::setup();
815        let ip_addr = IpAddr::V4(Ipv4Addr::UNSPECIFIED);
816        let (_server_port, (server_udp_socket, server_tcp_listener)) =
817            bind_common_in_range(ip_addr, (3200, 3250)).unwrap();
818
819        let _runtime = ip_echo_server(
820            server_tcp_listener,
821            DEFAULT_IP_ECHO_SERVER_THREADS,
822            /*shred_version=*/ Some(42),
823        );
824
825        let server_ip_echo_addr = server_udp_socket.local_addr().unwrap();
826        assert_eq!(
827            get_public_ip_addr(&server_ip_echo_addr),
828            parse_host("127.0.0.1"),
829        );
830        assert_eq!(get_cluster_shred_version(&server_ip_echo_addr), Ok(42));
831        assert!(verify_reachable_ports(&server_ip_echo_addr, vec![], &[],));
832    }
833
834    #[test]
835    fn test_get_public_ip_addr_reachable() {
836        solana_logger::setup();
837        let ip_addr = IpAddr::V4(Ipv4Addr::UNSPECIFIED);
838        let (_server_port, (server_udp_socket, server_tcp_listener)) =
839            bind_common_in_range(ip_addr, (3200, 3250)).unwrap();
840        let (client_port, (client_udp_socket, client_tcp_listener)) =
841            bind_common_in_range(ip_addr, (3200, 3250)).unwrap();
842
843        let _runtime = ip_echo_server(
844            server_tcp_listener,
845            DEFAULT_IP_ECHO_SERVER_THREADS,
846            /*shred_version=*/ Some(65535),
847        );
848
849        let ip_echo_server_addr = server_udp_socket.local_addr().unwrap();
850        assert_eq!(
851            get_public_ip_addr(&ip_echo_server_addr),
852            parse_host("127.0.0.1"),
853        );
854        assert_eq!(get_cluster_shred_version(&ip_echo_server_addr), Ok(65535));
855        assert!(verify_reachable_ports(
856            &ip_echo_server_addr,
857            vec![(client_port, client_tcp_listener)],
858            &[&client_udp_socket],
859        ));
860    }
861
862    #[test]
863    fn test_get_public_ip_addr_tcp_unreachable() {
864        solana_logger::setup();
865        let ip_addr = IpAddr::V4(Ipv4Addr::UNSPECIFIED);
866        let (_server_port, (server_udp_socket, _server_tcp_listener)) =
867            bind_common_in_range(ip_addr, (3200, 3250)).unwrap();
868
869        // make the socket unreachable by not running the ip echo server!
870
871        let server_ip_echo_addr = server_udp_socket.local_addr().unwrap();
872
873        let (correct_client_port, (_client_udp_socket, client_tcp_listener)) =
874            bind_common_in_range(ip_addr, (3200, 3250)).unwrap();
875
876        assert!(!do_verify_reachable_ports(
877            &server_ip_echo_addr,
878            vec![(correct_client_port, client_tcp_listener)],
879            &[],
880            2,
881            3,
882        ));
883    }
884
885    #[test]
886    fn test_get_public_ip_addr_udp_unreachable() {
887        solana_logger::setup();
888        let ip_addr = IpAddr::V4(Ipv4Addr::UNSPECIFIED);
889        let (_server_port, (server_udp_socket, _server_tcp_listener)) =
890            bind_common_in_range(ip_addr, (3200, 3250)).unwrap();
891
892        // make the socket unreachable by not running the ip echo server!
893
894        let server_ip_echo_addr = server_udp_socket.local_addr().unwrap();
895
896        let (_correct_client_port, (client_udp_socket, _client_tcp_listener)) =
897            bind_common_in_range(ip_addr, (3200, 3250)).unwrap();
898
899        assert!(!do_verify_reachable_ports(
900            &server_ip_echo_addr,
901            vec![],
902            &[&client_udp_socket],
903            2,
904            3,
905        ));
906    }
907
908    #[test]
909    fn test_bind_two_in_range_with_offset() {
910        solana_logger::setup();
911        let ip_addr = IpAddr::V4(Ipv4Addr::UNSPECIFIED);
912        let offset = 6;
913        if let Ok(((port1, _), (port2, _))) =
914            bind_two_in_range_with_offset(ip_addr, (1024, 65535), offset)
915        {
916            assert!(port2 == port1 + offset);
917        }
918        let offset = 42;
919        if let Ok(((port1, _), (port2, _))) =
920            bind_two_in_range_with_offset(ip_addr, (1024, 65535), offset)
921        {
922            assert!(port2 == port1 + offset);
923        }
924        assert!(bind_two_in_range_with_offset(ip_addr, (1024, 1044), offset).is_err());
925    }
926}
solana_net_utils/lib.rs

solana_net_utils/
lib.rs
