#[allow(dead_code)]
mod netns;
use ntkd::kernel::addressing;
use std::collections::HashMap;
use std::os::fd::{AsRawFd, RawFd};
use std::time::{Duration, Instant};
use anyhow::Context;
use futures::{StreamExt, TryStreamExt};
use ntk_common::{HCoord, Naddr, Topology};
use ntk_neighborhood::NodeId;
use ntk_netlink::Ipv4Net;
use tokio::task::JoinSet;
use tokio_util::sync::CancellationToken;
use wl_nl80211::packet_core::{DefaultNla, NLM_F_ACK, NLM_F_REQUEST};
use wl_nl80211::packet_generic::GenlMessage;
use wl_nl80211::{
Nl80211Attr, Nl80211BssInfo, Nl80211Command, Nl80211Error, Nl80211Handle, Nl80211Interface,
Nl80211InterfaceType, Nl80211Message,
};
fn topology() -> Topology {
Topology::new([4, 2, 2, 2]).unwrap()
}
fn negotiation_topology() -> Topology {
Topology::new([8]).unwrap()
}
fn position(idx: u32) -> Vec<u32> {
vec![idx, 0, 0, 0]
}
fn naddr(idx: u32) -> Naddr {
Naddr::new(topology(), position(idx)).unwrap()
}
const IBSS_SSID: &str = "ntkd-wireless-test";
const IBSS_FREQ_MHZ: u32 = 2412;
const IBSS_MERGE_TIMEOUT: Duration = Duration::from_secs(30);
const IBSS_FIXED_BSSID: [u8; 6] = [0x02, 0x4e, 0x54, 0x4b, 0x77, 0x69];
const IBSS_JOIN_CONFIRM_TIMEOUT: Duration = Duration::from_secs(10);
const NL80211_ATTR_FREQ_FIXED: u16 = 60;
const NL80211_ATTR_NETNS_FD: u16 = 219;
async fn nl80211_send(
handle: &mut Nl80211Handle,
cmd: Nl80211Command,
attributes: Vec<Nl80211Attr>,
) -> Result<(), Nl80211Error> {
let mut nl_msg =
wl_nl80211::packet_core::NetlinkMessage::from(GenlMessage::from_payload(Nl80211Message {
cmd,
attributes,
}));
nl_msg.header.flags = NLM_F_REQUEST | NLM_F_ACK;
let mut replies = std::pin::pin!(handle.request(nl_msg).await?);
while let Some(reply) = replies.next().await {
let _: GenlMessage<Nl80211Message> = wl_nl80211::try_nl80211!(reply);
}
Ok(())
}
async fn set_interface_adhoc(handle: &mut Nl80211Handle, if_index: u32) -> anyhow::Result<()> {
let attrs = Nl80211Interface::new(if_index)
.interface_type(Nl80211InterfaceType::Adhoc)
.build();
nl80211_send(handle, Nl80211Command::SetInterface, attrs).await?;
Ok(())
}
async fn join_ibss(
handle: &mut Nl80211Handle,
if_index: u32,
ssid: &str,
freq_mhz: u32,
bssid: Option<[u8; 6]>,
) -> anyhow::Result<()> {
let mut attrs = vec![
Nl80211Attr::IfIndex(if_index),
Nl80211Attr::Ssid(ssid.to_owned()),
Nl80211Attr::WiphyFreq(freq_mhz),
Nl80211Attr::Other(DefaultNla::new(NL80211_ATTR_FREQ_FIXED, Vec::new())),
];
if let Some(bssid) = bssid {
attrs.push(Nl80211Attr::Mac(bssid));
}
nl80211_send(handle, Nl80211Command::JoinIbss, attrs).await?;
Ok(())
}
async fn leave_ibss(handle: &mut Nl80211Handle, if_index: u32) -> anyhow::Result<()> {
nl80211_send(
handle,
Nl80211Command::LeaveIbss,
vec![Nl80211Attr::IfIndex(if_index)],
)
.await?;
Ok(())
}
const NL80211_BSS_STATUS_IBSS_JOINED: u32 = 2;
async fn read_joined_bssid(
handle: &mut Nl80211Handle,
if_index: u32,
) -> anyhow::Result<Option<[u8; 6]>> {
let mut replies = std::pin::pin!(handle.scan().dump(if_index).execute().await);
while let Some(msg) = replies.try_next().await? {
for attr in &msg.payload.attributes {
let Nl80211Attr::Bss(items) = attr else {
continue;
};
let joined = items.iter().any(
|i| matches!(i, Nl80211BssInfo::Status(s) if *s == NL80211_BSS_STATUS_IBSS_JOINED),
);
if !joined {
continue;
}
if let Some(Nl80211BssInfo::Bssid(mac)) =
items.iter().find(|i| matches!(i, Nl80211BssInfo::Bssid(_)))
{
return Ok(Some(*mac));
}
}
}
Ok(None)
}
async fn list_station_macs(
handle: &mut Nl80211Handle,
if_index: u32,
) -> anyhow::Result<Vec<[u8; 6]>> {
let mut macs = Vec::new();
let mut replies = std::pin::pin!(handle.station().dump(if_index).execute().await);
while let Some(msg) = replies.try_next().await? {
if let Some(Nl80211Attr::Mac(mac)) = msg
.payload
.attributes
.iter()
.find(|a| matches!(a, Nl80211Attr::Mac(_)))
{
macs.push(*mac);
}
}
Ok(macs)
}
fn fmt_mac(mac: &[u8; 6]) -> String {
mac.iter()
.map(|b| format!("{b:02x}"))
.collect::<Vec<_>>()
.join(":")
}
async fn set_wiphy_netns(
handle: &mut Nl80211Handle,
wiphy_index: u32,
netns_fd: RawFd,
) -> anyhow::Result<()> {
let attrs = vec![
Nl80211Attr::Wiphy(wiphy_index),
Nl80211Attr::Other(DefaultNla::new(
NL80211_ATTR_NETNS_FD,
(netns_fd as u32).to_ne_bytes().to_vec(),
)),
];
nl80211_send(handle, Nl80211Command::SetWiphyNetns, attrs).await?;
Ok(())
}
const HWSIM_DRIVER: &str = "mac80211_hwsim";
fn phy_driver(phy_name: &str) -> anyhow::Result<String> {
let link = format!("/sys/class/ieee80211/{phy_name}/device/driver");
let target = std::fs::read_link(&link)
.with_context(|| format!("reading driver symlink for {phy_name} ({link})"))?;
target
.file_name()
.and_then(|n| n.to_str())
.map(str::to_owned)
.ok_or_else(|| {
anyhow::anyhow!("driver symlink for {phy_name} has no file name: {target:?}")
})
}
#[derive(Debug, Clone)]
struct HwsimRadio {
phy_name: String,
wiphy_index: u32,
if_index: u32,
if_name: String,
}
async fn discover_hwsim_radios(handle: &mut Nl80211Handle) -> anyhow::Result<Vec<HwsimRadio>> {
let mut wiphy_names: HashMap<u32, String> = HashMap::new();
let mut wiphy_stream = std::pin::pin!(handle.wireless_physic().get().execute().await);
while let Some(msg) = wiphy_stream.try_next().await? {
let attrs = msg.payload.attributes;
let index = attrs.iter().find_map(|a| match a {
Nl80211Attr::Wiphy(i) => Some(*i),
_ => None,
});
let name = attrs.iter().find_map(|a| match a {
Nl80211Attr::WiphyName(n) => Some(n.clone()),
_ => None,
});
if let (Some(index), Some(name)) = (index, name) {
wiphy_names.entry(index).or_insert(name);
}
}
let mut radios = Vec::new();
let mut iface_stream = std::pin::pin!(handle.interface().get(vec![]).execute().await);
while let Some(msg) = iface_stream.try_next().await? {
let attrs = msg.payload.attributes;
let wiphy_index = attrs.iter().find_map(|a| match a {
Nl80211Attr::Wiphy(i) => Some(*i),
_ => None,
});
let if_index = attrs.iter().find_map(|a| match a {
Nl80211Attr::IfIndex(i) => Some(*i),
_ => None,
});
let if_name = attrs.iter().find_map(|a| match a {
Nl80211Attr::IfName(n) => Some(n.clone()),
_ => None,
});
let (Some(wiphy_index), Some(if_index), Some(if_name)) = (wiphy_index, if_index, if_name)
else {
continue;
};
let Some(phy_name) = wiphy_names.get(&wiphy_index).cloned() else {
continue;
};
if phy_driver(&phy_name)? == HWSIM_DRIVER {
radios.push(HwsimRadio {
phy_name,
wiphy_index,
if_index,
if_name,
});
}
}
radios.sort_by_key(|r| r.wiphy_index);
Ok(radios)
}
#[ignore = "needs mac80211_hwsim loaded (not root — see this test's own doc comment)"]
#[tokio::test]
async fn hwsim_radio_discovery_never_returns_the_real_radio() {
let (connection, mut handle, _) = wl_nl80211::new_connection().expect("nl80211 connection");
tokio::spawn(connection);
let radios = discover_hwsim_radios(&mut handle)
.await
.expect("discover hwsim radios");
assert!(
!radios.is_empty(),
"expected at least one mac80211_hwsim radio; is the module loaded \
(`modprobe mac80211_hwsim radios=2`)?"
);
for radio in &radios {
assert_ne!(
radio.phy_name, "phy0",
"phy0 is the host's real iwlwifi card and must never be classified as usable"
);
assert_ne!(
radio.if_name, "wlan0",
"wlan0 is the host's real iwlwifi card and must never be classified as usable"
);
assert_eq!(
phy_driver(&radio.phy_name).unwrap(),
HWSIM_DRIVER,
"{} was returned by discover_hwsim_radios but isn't actually hwsim-backed",
radio.phy_name
);
}
eprintln!(
"discovered {} hwsim radio(s): {:?}",
radios.len(),
radios.iter().map(|r| &r.phy_name).collect::<Vec<_>>()
);
}
async fn prepare_radio_interface(
if_name: &str,
ssid: &str,
freq_mhz: u32,
bssid: Option<[u8; 6]>,
label: &str,
) -> anyhow::Result<HashMap<String, u32>> {
let rt_handle =
netns::root_handle().with_context(|| format!("{label}: rtnetlink connection"))?;
netns::link::up(&rt_handle, "lo")
.await
.with_context(|| format!("{label}: bring lo up"))?;
let (nl_connection, mut nl_handle, _) =
wl_nl80211::new_connection().with_context(|| format!("{label}: nl80211 connection"))?;
tokio::spawn(nl_connection);
let deadline = tokio::time::Instant::now() + NETDEV_REAPPEAR_TIMEOUT;
let if_index = loop {
let attempt: anyhow::Result<u32> = async {
let if_index = netns::link::index(&rt_handle, if_name)
.await
.with_context(|| format!("{label}: resolve {if_name}'s ifindex after arrival"))?;
set_interface_adhoc(&mut nl_handle, if_index)
.await
.with_context(|| format!("{label}: set {if_name} to adhoc/IBSS type"))?;
Ok(if_index)
}
.await;
match attempt {
Ok(if_index) => break if_index,
Err(_) if tokio::time::Instant::now() < deadline => {
tokio::time::sleep(NETDEV_REAPPEAR_POLL_INTERVAL).await;
}
Err(e) => {
return Err(e.context(format!(
"{label}: {if_name} did not settle within {NETDEV_REAPPEAR_TIMEOUT:?}"
)));
}
}
};
let dev_index = netns::link::up(&rt_handle, if_name)
.await
.with_context(|| format!("{label}: bring {if_name} up"))?;
join_ibss(&mut nl_handle, if_index, ssid, freq_mhz, bssid)
.await
.with_context(|| format!("{label}: join IBSS {ssid:?} on {freq_mhz}MHz"))?;
let join_deadline = tokio::time::Instant::now() + IBSS_JOIN_CONFIRM_TIMEOUT;
loop {
let joined = read_joined_bssid(&mut nl_handle, if_index)
.await
.with_context(|| format!("{label}: read back {if_name}'s joined BSSID"))?;
match joined {
Some(actual) if bssid.is_none() || bssid == Some(actual) => break,
_ if tokio::time::Instant::now() >= join_deadline => {
return Err(anyhow::anyhow!(
"{label}: {if_name} did not report NL80211_BSS_STATUS_IBSS_JOINED for the \
requested cell within {IBSS_JOIN_CONFIRM_TIMEOUT:?} after JOIN_IBSS's own \
ack — joined={joined:?} wanted={bssid:?}"
));
}
_ => tokio::time::sleep(NETDEV_REAPPEAR_POLL_INTERVAL).await,
}
}
Ok(HashMap::from([(if_name.to_owned(), dev_index)]))
}
const NETDEV_REAPPEAR_TIMEOUT: Duration = Duration::from_secs(30);
const NETDEV_REAPPEAR_POLL_INTERVAL: Duration = Duration::from_millis(100);
const RADIO_RETURN_MOVE_TIMEOUT: Duration = Duration::from_secs(240);
const RADIO_RETURN_VERIFY_TIMEOUT: Duration = Duration::from_secs(240);
async fn wait_for_netdev(
rt_handle: &rtnetlink::Handle,
if_name: &str,
label: &str,
timeout: Duration,
) -> anyhow::Result<u32> {
let deadline = tokio::time::Instant::now() + timeout;
loop {
if let Ok(index) = netns::link::index(rt_handle, if_name).await {
return Ok(index);
}
if tokio::time::Instant::now() >= deadline {
return Err(anyhow::anyhow!(
"{label}: {if_name} did not resolve in this namespace within {timeout:?}"
));
}
tokio::time::sleep(NETDEV_REAPPEAR_POLL_INTERVAL).await;
}
}
async fn bring_down_and_move_wiphy(
rt_handle: &rtnetlink::Handle,
nl_handle: &mut Nl80211Handle,
if_name: &str,
wiphy_index: u32,
netns_fd: RawFd,
label: &str,
) -> anyhow::Result<()> {
let deadline = tokio::time::Instant::now() + NETDEV_REAPPEAR_TIMEOUT;
loop {
let attempt: anyhow::Result<()> = async {
let index = netns::link::index(rt_handle, if_name).await?;
netns::link::down(rt_handle, index)
.await
.with_context(|| format!("bringing {if_name} down before moving its wiphy"))?;
set_wiphy_netns(nl_handle, wiphy_index, netns_fd)
.await
.with_context(|| {
format!("moving wiphy {wiphy_index} (via {if_name}) to target namespace")
})
}
.await;
match attempt {
Ok(()) => return Ok(()),
Err(_) if tokio::time::Instant::now() < deadline => {
tokio::time::sleep(NETDEV_REAPPEAR_POLL_INTERVAL).await;
}
Err(e) => {
return Err(e.context(format!(
"{label}: {if_name} did not settle within {NETDEV_REAPPEAR_TIMEOUT:?}"
)));
}
}
}
}
async fn move_radio_into_namespace(
rt_handle: &rtnetlink::Handle,
nl_handle: &mut Nl80211Handle,
radio: &HwsimRadio,
netns_fd: RawFd,
) -> anyhow::Result<()> {
bring_down_and_move_wiphy(
rt_handle,
nl_handle,
&radio.if_name,
radio.wiphy_index,
netns_fd,
&radio.phy_name,
)
.await
.with_context(|| format!("moving {} into the target namespace", radio.phy_name))
}
fn open_init_net_fd(label: &str) -> anyhow::Result<std::fs::File> {
std::fs::File::open("/proc/self/ns/net")
.with_context(|| format!("{label}: open coordinator's own (init_net) netns fd"))
}
struct RadioReturnGuard {
label: String,
if_name: String,
wiphy_index: u32,
init_net: Option<std::fs::File>,
}
impl RadioReturnGuard {
fn new(label: &str, radio: &HwsimRadio, init_net: std::fs::File) -> Self {
Self {
label: label.to_owned(),
if_name: radio.if_name.clone(),
wiphy_index: radio.wiphy_index,
init_net: Some(init_net),
}
}
}
impl Drop for RadioReturnGuard {
fn drop(&mut self) {
let Some(init_net) = self.init_net.take() else {
return;
};
let label = self.label.clone();
let if_name = self.if_name.clone();
let wiphy_index = self.wiphy_index;
let spawned = std::thread::Builder::new()
.name(format!("{}-return", self.label))
.spawn(move || -> anyhow::Result<bool> {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.context("build return-to-init_net runtime")?;
rt.block_on(async move {
let rt_handle = netns::root_handle().context("rtnetlink connection")?;
let (nl_connection, mut nl_handle, _) =
wl_nl80211::new_connection().context("nl80211 connection")?;
tokio::spawn(nl_connection);
let move_deadline = tokio::time::Instant::now() + RADIO_RETURN_MOVE_TIMEOUT;
let move_error = loop {
match bring_down_and_move_wiphy(
&rt_handle,
&mut nl_handle,
&if_name,
wiphy_index,
init_net.as_raw_fd(),
&label,
)
.await
{
Ok(()) => break None,
Err(e) if tokio::time::Instant::now() < move_deadline => {
eprintln!(
"{label}: moving {if_name} back to init_net failed, \
retrying (budget {RADIO_RETURN_MOVE_TIMEOUT:?}): {e:#}"
);
}
Err(e) => break Some(e),
}
};
nix::sched::setns(&init_net, nix::sched::CloneFlags::CLONE_NEWNET)
.with_context(|| {
format!("{label}: setns back into init_net to verify {if_name}")
})?;
let verify_handle = netns::root_handle()
.with_context(|| format!("{label}: rtnetlink connection in init_net"))?;
let verified = wait_for_netdev(
&verify_handle,
&if_name,
&label,
RADIO_RETURN_VERIFY_TIMEOUT,
)
.await
.is_ok();
match (verified, move_error) {
(true, _) => Ok(true),
(false, None) => Ok(false),
(false, Some(e)) => Err(e.context(format!(
"{label}: moving {if_name} back to init_net within \
{RADIO_RETURN_MOVE_TIMEOUT:?}, and it never became resolvable in \
init_net within {RADIO_RETURN_VERIFY_TIMEOUT:?} either"
))),
}
})
});
let outcome = match spawned {
Ok(handle) => handle.join().unwrap_or_else(|e| {
Err(anyhow::anyhow!("return-to-init_net thread panicked: {e:?}"))
}),
Err(e) => Err(anyhow::anyhow!("spawn return-to-init_net thread: {e}")),
};
match outcome {
Ok(true) => {}
Ok(false) => eprintln!(
"{}: moved {}'s wiphy back to init_net but could not confirm it resolvable \
within {RADIO_RETURN_VERIFY_TIMEOUT:?} — NOT leaked (`init_net` was checked \
directly), likely still settling under NetworkManager/iwd contention",
self.label, self.if_name
),
Err(err) => eprintln!(
"{}: could not confirm {}'s wiphy returned to init_net within \
{RADIO_RETURN_MOVE_TIMEOUT:?} (move) + {RADIO_RETURN_VERIFY_TIMEOUT:?} \
(verify) — most likely still settling under NetworkManager/iwd \
contention, which has been measured taking minutes on this host. Re-check \
the pool after a delay before concluding anything was lost: {err:#}",
self.label, self.if_name
),
}
}
}
fn radio_worker_join_timeout() -> Duration {
IBSS_MERGE_TIMEOUT
+ netns::JOIN_MARGIN
+ NETDEV_REAPPEAR_TIMEOUT
+ IBSS_JOIN_CONFIRM_TIMEOUT
+ RADIO_RETURN_MOVE_TIMEOUT
+ RADIO_RETURN_VERIFY_TIMEOUT
}
#[derive(Debug)]
struct RadioTrialReport {
node: netns::NodeReport,
expected_destination: Ipv4Net,
route_installed: bool,
bssid: Option<[u8; 6]>,
stations: Vec<[u8; 6]>,
}
#[allow(clippy::too_many_arguments)]
async fn radio_arc_trial_body(
label: String,
my_id: NodeId,
my_idx: u32,
peer_idx: u32,
radio: HwsimRadio,
port: u16,
init_net: std::fs::File,
my_done_tx: tokio::sync::oneshot::Sender<()>,
peer_done_rx: tokio::sync::oneshot::Receiver<()>,
) -> anyhow::Result<RadioTrialReport> {
let _return_guard = RadioReturnGuard::new(&label, &radio, init_net);
let dev_index = prepare_radio_interface(
&radio.if_name,
IBSS_SSID,
IBSS_FREQ_MHZ,
Some(IBSS_FIXED_BSSID),
&label,
)
.await?;
let mut tasks = JoinSet::new();
let cancel = CancellationToken::new();
let started = netns::spawn_node(
my_id,
Some(position(my_idx)),
None,
&[4, 2, 2, 2],
&[radio.if_name.as_str()],
port,
&mut tasks,
cancel.clone(),
)
.await
.with_context(|| format!("{label}: compose real node"))?;
let expected_destination =
addressing::gnode_destination(&naddr(my_idx), HCoord::new(0, peer_idx))
.with_context(|| format!("{label}: expected destination"))?;
let deadline = tokio::time::Instant::now() + IBSS_MERGE_TIMEOUT;
let (node, route_installed) = loop {
let node = netns::observe(&label, &started, dev_index.clone())
.await
.with_context(|| format!("{label}: observe"))?;
let found = node
.routes
.iter()
.any(|r| r.destination == expected_destination);
if found || tokio::time::Instant::now() >= deadline {
break (node, found);
}
tokio::time::sleep(Duration::from_millis(200)).await;
};
let (bssid, stations) = match (
dev_index.get(radio.if_name.as_str()),
wl_nl80211::new_connection(),
) {
(Some(&idx), Ok((connection, mut measure_handle, _))) => {
tokio::spawn(connection);
let bssid = read_joined_bssid(&mut measure_handle, idx)
.await
.unwrap_or_else(|err| {
eprintln!("{label}: read_joined_bssid failed: {err:#}");
None
});
let stations = list_station_macs(&mut measure_handle, idx)
.await
.unwrap_or_else(|err| {
eprintln!("{label}: list_station_macs failed: {err:#}");
Vec::new()
});
(bssid, stations)
}
_ => (None, Vec::new()),
};
let _ = my_done_tx.send(());
let _ = peer_done_rx.await;
netns::teardown(&started, cancel, &mut tasks).await;
if let Ok((connection, mut leave_handle, _)) = wl_nl80211::new_connection() {
tokio::spawn(connection);
if let Some(&idx) = dev_index.get(radio.if_name.as_str()) {
let _ = leave_ibss(&mut leave_handle, idx).await;
}
}
Ok(RadioTrialReport {
node,
expected_destination,
route_installed,
bssid,
stations,
})
}
#[derive(Debug, Clone, Copy)]
struct ArcTrialOutcome {
arc_established: bool,
route_installed: bool,
elapsed: Duration,
}
async fn run_ibss_arc_trial(
rt_handle: &rtnetlink::Handle,
nl_handle: &mut Nl80211Handle,
radio_a: HwsimRadio,
radio_b: HwsimRadio,
port: u16,
) -> anyhow::Result<(ArcTrialOutcome, RadioTrialReport, RadioTrialReport)> {
let started_at = Instant::now();
let move_a = radio_a.clone();
let move_b = radio_b.clone();
let init_net_a = open_init_net_fd("radio-a")?;
let init_net_b = open_init_net_fd("radio-b")?;
let (done_tx_a, done_rx_a) = tokio::sync::oneshot::channel::<()>();
let (done_tx_b, done_rx_b) = tokio::sync::oneshot::channel::<()>();
let worker_a = netns::NamespaceWorker::spawn("radio-a", move || {
radio_arc_trial_body(
"radio-a".to_owned(),
NodeId::from_raw(301).unwrap(),
0,
1,
radio_a,
port,
init_net_a,
done_tx_a,
done_rx_b,
)
});
let worker_b = netns::NamespaceWorker::spawn("radio-b", move || {
radio_arc_trial_body(
"radio-b".to_owned(),
NodeId::from_raw(302).unwrap(),
1,
0,
radio_b,
port,
init_net_b,
done_tx_b,
done_rx_a,
)
});
let fd_a = worker_a.fd();
let fd_b = worker_b.fd();
move_radio_into_namespace(rt_handle, nl_handle, &move_a, fd_a).await?;
move_radio_into_namespace(rt_handle, nl_handle, &move_b, fd_b).await?;
worker_a.signal_moved();
worker_b.signal_moved();
let report_a = worker_a
.join(radio_worker_join_timeout())
.await
.context("radio-a namespace body")?;
let report_b = worker_b
.join(radio_worker_join_timeout())
.await
.context("radio-b namespace body")?;
let outcome = ArcTrialOutcome {
arc_established: report_a.node.arcs.iter().any(|a| a.cost.is_some())
&& report_b.node.arcs.iter().any(|a| a.cost.is_some()),
route_installed: report_a.route_installed && report_b.route_installed,
elapsed: started_at.elapsed(),
};
Ok((outcome, report_a, report_b))
}
#[ignore = "requires real root (CAP_NET_ADMIN in init_user_ns) — see this test's own doc comment"]
#[tokio::test]
async fn real_hwsim_two_daemons_establish_arc_and_route_over_ibss() {
let filter = tracing_subscriber::EnvFilter::try_from_default_env()
.unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("debug"));
let _ = tracing_subscriber::fmt()
.with_env_filter(filter)
.with_test_writer()
.try_init();
let rt_handle = netns::root_handle().expect("rtnetlink connection");
let (nl_connection, mut nl_handle, _) =
wl_nl80211::new_connection().expect("nl80211 connection");
tokio::spawn(nl_connection);
let radios = discover_hwsim_radios(&mut nl_handle)
.await
.expect("discover hwsim radios");
assert!(
radios.len() >= 2,
"need at least 2 mac80211_hwsim radios, found {}",
radios.len()
);
let (radio_a, radio_b) = (radios[0].clone(), radios[1].clone());
let (outcome, report_a, report_b) =
run_ibss_arc_trial(&rt_handle, &mut nl_handle, radio_a, radio_b, 27290)
.await
.expect("run ibss arc trial");
eprintln!(
"{}: bssid={} stations={:?} arcs={:#?} routes={:#?}",
report_a.node.label,
report_a.bssid.as_ref().map_or("none".to_owned(), fmt_mac),
report_a.stations.iter().map(fmt_mac).collect::<Vec<_>>(),
report_a.node.arcs,
report_a.node.routes
);
eprintln!(
"{}: bssid={} stations={:?} arcs={:#?} routes={:#?}",
report_b.node.label,
report_b.bssid.as_ref().map_or("none".to_owned(), fmt_mac),
report_b.stations.iter().map(fmt_mac).collect::<Vec<_>>(),
report_b.node.arcs,
report_b.node.routes
);
assert!(
report_a.node.arcs.iter().any(|a| a.cost.is_some()),
"radio-a never measured a cost for its neighbor — no arc established: {:#?}",
report_a.node.arcs
);
assert!(
report_b.node.arcs.iter().any(|a| a.cost.is_some()),
"radio-b never measured a cost for its neighbor — no arc established: {:#?}",
report_b.node.arcs
);
assert!(
report_a.route_installed,
"radio-a's real kernel routing table never gained a route to {}: routes={:#?} \
addresses={:#?}",
report_a.expected_destination, report_a.node.routes, report_a.node.addresses
);
assert!(
report_b.route_installed,
"radio-b's real kernel routing table never gained a route to {}: routes={:#?} \
addresses={:#?}",
report_b.expected_destination, report_b.node.routes, report_b.node.addresses
);
assert!(outcome.arc_established && outcome.route_installed);
}
#[allow(clippy::too_many_arguments)]
async fn radio_negotiation_trial_body(
label: String,
my_id: NodeId,
radio: HwsimRadio,
port: u16,
init_net: std::fs::File,
my_done_tx: tokio::sync::oneshot::Sender<()>,
peer_done_rx: tokio::sync::oneshot::Receiver<()>,
) -> anyhow::Result<(netns::NodeReport, Vec<u32>)> {
let _return_guard = RadioReturnGuard::new(&label, &radio, init_net);
let dev_index = prepare_radio_interface(
&radio.if_name,
IBSS_SSID,
IBSS_FREQ_MHZ,
Some(IBSS_FIXED_BSSID),
&label,
)
.await?;
let mut tasks = JoinSet::new();
let cancel = CancellationToken::new();
let started = netns::spawn_node(
my_id,
None,
None,
&[8],
&[radio.if_name.as_str()],
port,
&mut tasks,
cancel.clone(),
)
.await
.with_context(|| format!("{label}: compose real negotiating node"))?;
let initial_position = started
.running
.generation
.borrow()
.qspn
.my_naddr()
.positions()
.to_vec();
let deadline = tokio::time::Instant::now() + IBSS_MERGE_TIMEOUT;
let node = loop {
let node = netns::observe(&label, &started, dev_index.clone())
.await
.with_context(|| format!("{label}: observe"))?;
let rehooked_and_reattached = node.rehooked && !node.routes.is_empty();
if rehooked_and_reattached || tokio::time::Instant::now() >= deadline {
break node;
}
tokio::time::sleep(Duration::from_millis(200)).await;
};
let _ = my_done_tx.send(());
let _ = peer_done_rx.await;
netns::teardown(&started, cancel, &mut tasks).await;
if let Ok((connection, mut leave_handle, _)) = wl_nl80211::new_connection() {
tokio::spawn(connection);
if let Some(&idx) = dev_index.get(radio.if_name.as_str()) {
let _ = leave_ibss(&mut leave_handle, idx).await;
}
}
Ok((node, initial_position))
}
#[ignore = "requires real root (CAP_NET_ADMIN in init_user_ns) — see this test's own doc comment"]
#[tokio::test]
async fn real_hwsim_two_virgin_daemons_negotiate_over_the_air() {
let filter = tracing_subscriber::EnvFilter::try_from_default_env()
.unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("debug"));
let _ = tracing_subscriber::fmt()
.with_env_filter(filter)
.with_test_writer()
.with_thread_names(true)
.try_init();
let rt_handle = netns::root_handle().expect("rtnetlink connection");
let (nl_connection, mut nl_handle, _) =
wl_nl80211::new_connection().expect("nl80211 connection");
tokio::spawn(nl_connection);
let radios = discover_hwsim_radios(&mut nl_handle)
.await
.expect("discover hwsim radios");
assert!(
radios.len() >= 2,
"need at least 2 mac80211_hwsim radios, found {}",
radios.len()
);
let (radio_a, radio_b) = (radios[0].clone(), radios[1].clone());
const PORT: u16 = 27291;
let move_a = radio_a.clone();
let move_b = radio_b.clone();
let init_net_a = open_init_net_fd("radio-a").expect("open coordinator init_net fd for radio-a");
let init_net_b = open_init_net_fd("radio-b").expect("open coordinator init_net fd for radio-b");
let (done_tx_a, done_rx_a) = tokio::sync::oneshot::channel::<()>();
let (done_tx_b, done_rx_b) = tokio::sync::oneshot::channel::<()>();
let worker_a = netns::NamespaceWorker::spawn("radio-a", move || {
radio_negotiation_trial_body(
"radio-a".to_owned(),
NodeId::from_raw(401).unwrap(),
radio_a,
PORT,
init_net_a,
done_tx_a,
done_rx_b,
)
});
let worker_b = netns::NamespaceWorker::spawn("radio-b", move || {
radio_negotiation_trial_body(
"radio-b".to_owned(),
NodeId::from_raw(402).unwrap(),
radio_b,
PORT,
init_net_b,
done_tx_b,
done_rx_a,
)
});
let fd_a = worker_a.fd();
let fd_b = worker_b.fd();
move_radio_into_namespace(&rt_handle, &mut nl_handle, &move_a, fd_a)
.await
.expect("move radio-a's wiphy into ns-a");
move_radio_into_namespace(&rt_handle, &mut nl_handle, &move_b, fd_b)
.await
.expect("move radio-b's wiphy into ns-b");
worker_a.signal_moved();
worker_b.signal_moved();
let (report_a, initial_a) = worker_a
.join(radio_worker_join_timeout())
.await
.expect("radio-a namespace body");
let (report_b, initial_b) = worker_b
.join(radio_worker_join_timeout())
.await
.expect("radio-b namespace body");
eprintln!("{}: {report_a:#?} initial={initial_a:?}", report_a.label);
eprintln!("{}: {report_b:#?} initial={initial_b:?}", report_b.label);
assert_ne!(
report_a.rehooked, report_b.rehooked,
"exactly one side should adopt the negotiated position, never both or neither"
);
let (winner, loser, loser_initial) = if report_a.rehooked {
(&report_b, &report_a, &initial_a)
} else {
(&report_a, &report_b, &initial_b)
};
assert_eq!(
loser.naddr_positions.len(),
1,
"single-level topology: a resolved position always has exactly one entry"
);
assert_ne!(
&loser.naddr_positions,
if report_a.rehooked {
&initial_b
} else {
&initial_a
},
"the loser must resolve to a free slot in the winner's network, not the winner's own \
already-occupied position"
);
let topology = negotiation_topology();
let old_address =
addressing::host_address(&Naddr::new(topology.clone(), loser_initial.clone()).unwrap())
.expect("loser's own trivial address is always addressable");
let new_address =
addressing::host_address(&Naddr::new(topology, loser.naddr_positions.clone()).unwrap())
.expect("loser's negotiated address is always addressable");
let netsukuku_addresses: Vec<_> = loser
.addresses
.iter()
.filter(|a| a.network.address().octets()[0] == 10)
.collect();
assert_eq!(
netsukuku_addresses.len(),
1,
"the loser's real kernel address table should carry exactly one Netsukuku-space \
address (its negotiated one, {new_address}) — a second entry would mean its \
torn-down trivial-generation address {old_address} leaked alongside it: {:#?}",
loser.addresses
);
assert_eq!(
netsukuku_addresses[0].network, new_address,
"the loser's sole Netsukuku-space address is not its negotiated one: {:#?}",
loser.addresses
);
assert!(
!winner.routes.is_empty(),
"the winner's real kernel routing table never gained a route to its (migrated) peer: \
{:#?}",
winner.routes
);
assert!(
!loser.routes.is_empty(),
"the loser's real kernel routing table never gained a route to its (winner) peer after \
rehook re-attached the arc: {:#?}",
loser.routes
);
}
#[derive(Debug)]
struct ReliabilitySummary {
attempts: usize,
successes: usize,
time_to_arc: Vec<Duration>,
}
impl ReliabilitySummary {
fn report(&self) -> String {
if self.time_to_arc.is_empty() {
return format!(
"{}/{} runs formed an arc within the deadline; no successful run to time",
self.successes, self.attempts
);
}
let mut sorted = self.time_to_arc.clone();
sorted.sort();
let min = sorted.first().copied().unwrap();
let max = sorted.last().copied().unwrap();
let median = sorted[sorted.len() / 2];
let total: Duration = sorted.iter().sum();
let mean = total / sorted.len() as u32;
format!(
"{}/{} runs formed an arc within the deadline; time-to-arc over the {} successes: \
min={min:?} median={median:?} mean={mean:?} max={max:?}",
self.successes,
self.attempts,
sorted.len()
)
}
}
#[ignore = "requires real root (CAP_NET_ADMIN in init_user_ns) — see this test's own doc comment"]
#[tokio::test]
async fn real_hwsim_broadcast_reliability_across_ten_runs() {
let filter = tracing_subscriber::EnvFilter::try_from_default_env()
.unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("debug"));
let _ = tracing_subscriber::fmt()
.with_env_filter(filter)
.with_test_writer()
.with_thread_names(true)
.try_init();
const RUNS: usize = 10;
let rt_handle = netns::root_handle().expect("rtnetlink connection");
let (nl_connection, mut nl_handle, _) =
wl_nl80211::new_connection().expect("nl80211 connection");
tokio::spawn(nl_connection);
let radios = discover_hwsim_radios(&mut nl_handle)
.await
.expect("discover hwsim radios");
assert!(
radios.len() >= 2,
"need at least 2 mac80211_hwsim radios, found {}",
radios.len()
);
let (radio_a, radio_b) = (radios[0].clone(), radios[1].clone());
let mut successes = 0usize;
let mut time_to_arc = Vec::new();
for run in 0..RUNS {
let port = 27300 + (run as u16) * 2;
let trial = run_ibss_arc_trial(
&rt_handle,
&mut nl_handle,
radio_a.clone(),
radio_b.clone(),
port,
)
.await;
match trial {
Ok((outcome, _, _)) if outcome.arc_established && outcome.route_installed => {
successes += 1;
time_to_arc.push(outcome.elapsed);
eprintln!("run {run}: arc established in {:?}", outcome.elapsed);
}
Ok((outcome, report_a, report_b)) => {
eprintln!(
"run {run}: no arc within {:?} — a_bssid={} b_bssid={} a={:?} b={:?}",
outcome.elapsed,
report_a.bssid.as_ref().map_or("none".to_owned(), fmt_mac),
report_b.bssid.as_ref().map_or("none".to_owned(), fmt_mac),
report_a.node.arcs,
report_b.node.arcs
);
}
Err(err) => {
eprintln!("run {run}: trial setup failed: {err:#}");
}
}
}
let summary = ReliabilitySummary {
attempts: RUNS,
successes,
time_to_arc,
};
eprintln!("{}", summary.report());
}
#[ignore = "requires real root (CAP_NET_ADMIN) — see this test's own doc comment"]
#[tokio::test]
async fn hwsim_single_radio_join_ibss_smoke_test() {
let rt_handle = netns::root_handle().expect("rtnetlink connection");
let (nl_connection, mut nl_handle, _) =
wl_nl80211::new_connection().expect("nl80211 connection");
tokio::spawn(nl_connection);
let radios = discover_hwsim_radios(&mut nl_handle)
.await
.expect("discover hwsim radios");
let radio = radios.first().expect("at least one hwsim radio").clone();
assert_ne!(radio.phy_name, "phy0");
assert_ne!(radio.if_name, "wlan0");
set_interface_adhoc(&mut nl_handle, radio.if_index)
.await
.expect("set interface to adhoc/IBSS type");
netns::link::up(&rt_handle, &radio.if_name)
.await
.expect("bring radio up");
join_ibss(
&mut nl_handle,
radio.if_index,
IBSS_SSID,
IBSS_FREQ_MHZ,
Some(IBSS_FIXED_BSSID),
)
.await
.expect("join IBSS");
leave_ibss(&mut nl_handle, radio.if_index)
.await
.expect("leave IBSS");
let index = netns::link::index(&rt_handle, &radio.if_name)
.await
.expect("resolve radio index");
netns::link::down(&rt_handle, index)
.await
.expect("bring radio back down");
}
const RETURN_GUARD_FAILURE_TRIALS: usize = 6;
async fn deliberately_failing_trial_body(
label: String,
radio: HwsimRadio,
init_net: std::fs::File,
) -> anyhow::Result<()> {
let _return_guard = RadioReturnGuard::new(&label, &radio, init_net);
prepare_radio_interface(
&radio.if_name,
IBSS_SSID,
IBSS_FREQ_MHZ,
Some(IBSS_FIXED_BSSID),
&label,
)
.await?;
anyhow::bail!("{label}: deliberate failure to exercise RadioReturnGuard's failure path")
}
#[ignore = "requires real root (CAP_NET_ADMIN in init_user_ns) — see this test's own doc comment"]
#[tokio::test]
async fn radio_return_guard_survives_a_deliberately_failing_trial() {
let rt_handle = netns::root_handle().expect("rtnetlink connection");
let (nl_connection, mut nl_handle, _) =
wl_nl80211::new_connection().expect("nl80211 connection");
tokio::spawn(nl_connection);
let radios = discover_hwsim_radios(&mut nl_handle)
.await
.expect("discover hwsim radios");
let radio = radios.last().expect("at least one hwsim radio").clone();
eprintln!(
"using {} ({}) for {RETURN_GUARD_FAILURE_TRIALS} deliberately-failing trials",
radio.phy_name, radio.if_name
);
for trial in 0..RETURN_GUARD_FAILURE_TRIALS {
let started = Instant::now();
let init_net = open_init_net_fd("return-guard-fail").expect("open coordinator init_net fd");
let label = format!("return-guard-fail-{trial}");
let worker_radio = radio.clone();
let worker_label = label.clone();
let worker = netns::NamespaceWorker::spawn(label.clone(), move || {
deliberately_failing_trial_body(worker_label, worker_radio, init_net)
});
let fd = worker.fd();
move_radio_into_namespace(&rt_handle, &mut nl_handle, &radio, fd)
.await
.unwrap_or_else(|e| panic!("trial {trial}: move radio into namespace: {e:#}"));
worker.signal_moved();
let outcome = worker.join(radio_worker_join_timeout()).await;
let elapsed = started.elapsed();
assert!(
outcome.is_err(),
"trial {trial}: body was supposed to fail deliberately, but it returned Ok"
);
let poll_deadline = tokio::time::Instant::now() + Duration::from_secs(10);
let still_present = loop {
let after = discover_hwsim_radios(&mut nl_handle)
.await
.expect("discover hwsim radios after trial");
if after.iter().any(|r| r.phy_name == radio.phy_name) {
break true;
}
if tokio::time::Instant::now() >= poll_deadline {
break false;
}
tokio::time::sleep(NETDEV_REAPPEAR_POLL_INTERVAL).await;
};
eprintln!("trial {trial}: elapsed={elapsed:?} radio_returned={still_present}");
assert!(
still_present,
"trial {trial}: {} did not reappear in init_net (via discover_hwsim_radios) within \
10s of a deliberately-failing trial completing — RadioReturnGuard leaked it \
(trial elapsed {elapsed:?})",
radio.phy_name
);
}
}