test(server): verify wildcard source selection

crates/irtt-server/tests/wildcard_source.rs

@@ -0,0 +1,214 @@
+//! A wildcard listener answers from the address the request was sent to.
+//!
+//! Every test here drives a real [`AsyncClient`], which reads from a *connected*
+//! UDP socket. That connection is the assertion: the kernel discards any
+//! datagram whose source is not the endpoint the client sent to, so a reply
+//! that left from another of the host's addresses is indistinguishable from no
+//! reply at all. An open, an echo and a close completing therefore prove the
+//! source address, without the test having to inspect it.
+//!
+//! Compiled on the targets that implement wildcard source selection. Elsewhere
+//! a wildcard bind is refused outright, which the runtime's own tests cover.
+
+#![cfg(any(target_os = "linux", target_os = "macos", target_os = "freebsd"))]
+
+use std::{
+    io,
+    net::{Ipv4Addr, SocketAddr, UdpSocket},
+    time::Duration,
+};
+
+use irtt_client::{AsyncClient, ClientConfig, ClientEvent, OpenOutcome, PacketMeta, SocketConfig};
+use irtt_proto::{Clock, ReceivedStats, StampAt};
+use irtt_server::{Server, ServerConfig, ServerRuntimeError};
+use tokio::{sync::oneshot, task::JoinHandle, time::timeout};
+
+/// EF, as the client's configuration takes it, and the raw TOS byte it occupies
+/// the upper six bits of.
+const EF_DSCP: u8 = 46;
+const EF_TRAFFIC_CLASS: u8 = 0xb8;
+
+#[tokio::test(flavor = "current_thread")]
+async fn a_wildcard_ipv4_listener_replies_from_the_requested_address() {
+    timeout(Duration::from_secs(5), async {
+        let server = Server::bind("0.0.0.0:0".parse().unwrap(), unthrottled())
+            .await
+            .expect("a wildcard IPv4 listener must be supported on this target");
+        let port = server.local_addr().unwrap().port();
+        // A second loopback address makes this a genuine multi-homed case: the
+        // listener is bound to none of the host's addresses in particular, and
+        // the client contacts one the routing table would not have picked on
+        // its own. Hosts carrying only 127.0.0.1 — macOS, unless an alias is
+        // configured — still exercise the whole ancillary path, one address
+        // wide.
+        let target = SocketAddr::from((secondary_ipv4_loopback(), port));
+        let running = spawn(server);
+
+        let mut client = connected_client(target).await;
+        assert_started(&mut client, target).await;
+
+        let meta = echo_once(&mut client).await;
+        // Source selection is per-packet ancillary data and the traffic class
+        // is a socket-wide setting applied before the send. Both act on the
+        // same reply, so this is where they have to agree.
+        if cfg!(target_os = "linux") {
+            assert_eq!(
+                meta.traffic_class,
+                Some(EF_TRAFFIC_CLASS),
+                "a source-selected reply still carries its session's marking"
+            );
+            assert_eq!(meta.dscp, Some(EF_DSCP));
+            assert_eq!(meta.ecn, Some(0));
+        }
+
+        assert!(matches!(
+            client.close().await.unwrap().as_slice(),
+            [ClientEvent::SessionClosed { .. }]
+        ));
+        running.stop().await;
+    })
+    .await
+    .expect("the wildcard IPv4 source-selection test exceeded its bounded runtime");
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn a_wildcard_ipv6_listener_replies_from_the_requested_address() {
+    timeout(Duration::from_secs(5), async {
+        if !ipv6_loopback_available() {
+            eprintln!("skipping wildcard IPv6 test: IPv6 loopback unavailable");
+            return;
+        }
+        let server = match Server::bind("[::]:0".parse().unwrap(), unthrottled()).await {
+            Ok(server) => server,
+            Err(ServerRuntimeError::Bind { source, .. }) if is_family_unavailable(&source) => {
+                eprintln!("skipping wildcard IPv6 test: IPv6 unavailable");
+                return;
+            }
+            Err(error) => panic!("unexpected wildcard IPv6 bind failure: {error}"),
+        };
+        let port = server.local_addr().unwrap().port();
+        let target: SocketAddr = format!("[::1]:{port}").parse().unwrap();
+        let running = spawn(server);
+
+        // Loopback is one address, so this does not prove multi-homed IPv6. It
+        // does exercise the parts that differ from IPv4: the received packet
+        // info, the interface index carried with it, and the source-selected
+        // IPv6 send.
+        let mut client = connected_client(target).await;
+        assert_started(&mut client, target).await;
+        echo_once(&mut client).await;
+        assert!(matches!(
+            client.close().await.unwrap().as_slice(),
+            [ClientEvent::SessionClosed { .. }]
+        ));
+
+        running.stop().await;
+    })
+    .await
+    .expect("the wildcard IPv6 source-selection test exceeded its bounded runtime");
+}
+
+/// A local IPv4 address that is not the primary loopback one, where the host
+/// has one.
+///
+/// Bindability is the test: an address this process can bind is one the host
+/// accepts datagrams for. `127.0.0.0/8` supplies a second local destination on
+/// hosts that carry the whole range, without touching interface configuration.
+fn secondary_ipv4_loopback() -> Ipv4Addr {
+    let candidate = Ipv4Addr::new(127, 0, 0, 2);
+    if UdpSocket::bind((candidate, 0)).is_ok() {
+        return candidate;
+    }
+    eprintln!("{candidate} is not a local address on this host; using 127.0.0.1 instead");
+    Ipv4Addr::LOCALHOST
+}
+
+fn ipv6_loopback_available() -> bool {
+    match UdpSocket::bind("[::1]:0") {
+        Ok(_) => true,
+        Err(error) if is_family_unavailable(&error) => false,
+        Err(error) => panic!("unexpected IPv6 loopback bind failure: {error}"),
+    }
+}
+
+fn is_family_unavailable(error: &io::Error) -> bool {
+    matches!(
+        error.kind(),
+        io::ErrorKind::AddrNotAvailable | io::ErrorKind::Unsupported
+    )
+}
+
+fn unthrottled() -> ServerConfig {
+    ServerConfig::default().with_min_send_interval(Duration::ZERO)
+}
+
+async fn connected_client(server_addr: SocketAddr) -> AsyncClient {
+    let config = ClientConfig {
+        server_addr: server_addr.to_string(),
+        duration: Some(Duration::from_secs(30)),
+        interval: Duration::from_millis(10),
+        length: 64,
+        dscp: EF_DSCP,
+        received_stats: ReceivedStats::Both,
+        stamp_at: StampAt::Both,
+        clock: Clock::Both,
+        open_timeouts: vec![Duration::from_millis(500)],
+        socket_config: SocketConfig {
+            recv_timeout: Some(Duration::from_millis(500)),
+            ..SocketConfig::default()
+        },
+        probe_timeout: Duration::from_millis(500),
+        ..ClientConfig::default()
+    };
+    AsyncClient::connect(config).await.unwrap()
+}
+
+async fn assert_started(client: &mut AsyncClient, server_addr: SocketAddr) {
+    match client.open().await.unwrap() {
+        OpenOutcome::Started { remote, token, .. } => {
+            assert_eq!(remote, server_addr);
+            assert_ne!(token, 0);
+        }
+        OpenOutcome::NoTestCompleted { .. } => panic!("a normal client unexpectedly ran no-test"),
+    }
+}
+
+/// Sends one echo and returns its reply's receive metadata.
+async fn echo_once(client: &mut AsyncClient) -> PacketMeta {
+    client.send_probe().await.unwrap();
+    loop {
+        for event in client.recv().await.unwrap() {
+            if let ClientEvent::EchoReply {
+                seq, packet_meta, ..
+            } = event
+            {
+                assert_eq!(seq, 0);
+                return packet_meta;
+            }
+        }
+    }
+}
+
+struct Running {
+    shutdown: oneshot::Sender<()>,
+    task: JoinHandle<Result<(), ServerRuntimeError>>,
+}
+
+impl Running {
+    async fn stop(self) {
+        self.shutdown.send(()).unwrap();
+        self.task.await.unwrap().unwrap();
+    }
+}
+
+fn spawn(mut server: Server) -> Running {
+    let (shutdown, shutdown_rx) = oneshot::channel();
+    let task = tokio::spawn(async move {
+        server
+            .run(async {
+                let _ = shutdown_rx.await;
+            })
+            .await
+    });
+    Running { shutdown, task }
+}