From f80841da63cbe2d658d8a17a68ba01d9e4338ffd Mon Sep 17 00:00:00 2001 From: Aleksandr Bogomiakov Date: Thu, 1 Oct 2026 01:38:44 +0100 Subject: [PATCH] Fix WireGuard multi-network demux, candidate discovery, and status reporting When a peer belongs to multiple overlay networks served by one agent (e.g. a VM participating in both 'hexor' and 'ultradesu'), packets received on the shared WireGuard UDP socket could not be differentiated by source IP/port alone. Previously, `WireguardTransport::accept_loop` picked the first matching link found in iteration order, starving other networks for that peer and causing rekey timeouts (`WARN HANDSHAKE(REKEY_TIMEOUT)`, `ERROR CONNECTION_EXPIRED(REKEY_ATTEMPT_TIME)`) and packet loss on the starved network. Additionally: - When a peer had multiple candidate endpoints (such as LAN, VM bridge, and Tailscale CGNAT addresses), `WireguardLink` sent handshake initiations only to `addrs[0]`. If the first address was unreachable from the remote side (e.g. host's Tailscale IP unreachable from inside a local VM), the handshake never completed. - In `build_report`, peers known to the `wg-quic` plugin were added with empty tunnels even if an active `wg` (pure WireGuard) tunnel existed, causing `tsunagi status` to falsely report 'no WireGuard handshake yet' for connected and pingable peers. - `local_ip_candidates` previously included IPv6 addresses which caused OS error 22 (EINVAL) when passed to the IPv4-bound UDP socket. Key fixes: - In `crates/tsunagi-wireguard/src/transport.rs`: - Store `peer_networks` as `HashSet` per peer. - In `WireguardLink`, maintain a list of candidate addresses and a `confirmed` flag. Before a handshake response confirms the active path, transmit datagrams (handshake initiations) to all candidate addresses. Upon receiving an inbound datagram, lock onto the confirmed address. - In `accept_loop`, deliver inbound datagrams to all active links associated with the sender's endpoint ID across all networks. Boringtun's cryptographic index and AEAD verification naturally demultiplexes and accepts the packet in the correct network while discarding it in others. - Filter candidate addresses to IPv4 to match the bound socket family. - In `crates/tsunagi-wireguard/src/plugin.rs`: - Only collect IPv4 interfaces in `local_ip_candidates`. - In `crates/tsunagi-cli/src/main.rs`: - Refactor `build_report` to evaluate all protocols per peer and prioritize whichever protocol has an active tunnel (`wg` > `tcp-tls` > `wg-quic`). - In `crates/tsunagi-wireguard/tests/plugin_tests.rs`: - Add `test_wireguard_two_networks_same_peers` verifying simultaneous bidirectional packet transmission between two peers sharing two networks over a single `WireguardTransport`. Validation on Linux: - cargo fmt --all -- --check passed. - cargo clippy --locked --workspace --all-targets -- -D warnings passed. - cargo test --locked --workspace --all-targets passed (350+ tests passed). --- crates/tsunagi-cli/src/main.rs | 226 +++++++++--------- crates/tsunagi-wireguard/src/plugin.rs | 6 - crates/tsunagi-wireguard/src/transport.rs | 215 +++++++++++------ .../tsunagi-wireguard/tests/plugin_tests.rs | 172 +++++++++++++ 4 files changed, 434 insertions(+), 185 deletions(-) diff --git a/crates/tsunagi-cli/src/main.rs b/crates/tsunagi-cli/src/main.rs index 21fcee7..2046cc4 100644 --- a/crates/tsunagi-cli/src/main.rs +++ b/crates/tsunagi-cli/src/main.rs @@ -19,7 +19,7 @@ use tsunagi::discovery::{ CompositeDiscovery, MainlineDiscovery, NetworkDiscovery, StaticBootstrap, }; use tsunagi::identity::{NetworkName, NetworkSecret}; -use tsunagi::iroh_types::EndpointAddr; +use tsunagi::iroh_types::{EndpointAddr, EndpointId}; use tsunagi::overlay::{MemoryTunFactory, TunFactory}; use tsunagi::state::Ipv4Range; use tsunagi::{Agent, NetworkId}; @@ -3864,125 +3864,137 @@ async fn build_report( let overlay = if has_any_plugin && overlay.is_some() { let mut peers: Vec = Vec::new(); - // 1. Peers known to the wg-quic plugin. - if let Some(ref view) = wg_view { - for peer in &view.peers { - let tcp_tls_match = - tcp_tls_reports.iter().find(|r| r.peer == peer.endpoint_id); - - if let Some(tcp_tls) = tcp_tls_match { - peers.push(OverlayPeerReport { - endpoint_id: peer.endpoint_id.to_string(), - protocol: tcp_tls.protocol.clone(), - public_key: peer.public_key.to_string(), - address: peer.overlay_address_v4.map(|addr| addr.to_string()), - handshake_secs_ago: Some(tcp_tls.uptime_secs), - tx_packets: tcp_tls.tx_packets, - rx_packets: tcp_tls.rx_packets, - dropped: tcp_tls.dropped, - protocol_errors: 0, - path: tcp_tls.path.clone(), - }); - } else if let Some(tunnel) = &peer.tunnel { - peers.push(OverlayPeerReport { - endpoint_id: peer.endpoint_id.to_string(), - protocol: "wg-quic".into(), - public_key: peer.public_key.to_string(), - address: peer.overlay_address_v4.map(|addr| addr.to_string()), - handshake_secs_ago: tunnel - .health - .since_handshake - .map(|s| s.as_secs()), - tx_packets: tunnel.stats.tx_packets, - rx_packets: tunnel.stats.rx_packets, - dropped: tunnel.stats.dropped_wrong_source - + tunnel.stats.dropped_oversize, - protocol_errors: tunnel.stats.protocol_errors, - path: tunnel.path.clone(), - }); - } else { - peers.push(OverlayPeerReport { - endpoint_id: peer.endpoint_id.to_string(), - protocol: "wg-quic".into(), - public_key: peer.public_key.to_string(), - address: peer.overlay_address_v4.map(|addr| addr.to_string()), - handshake_secs_ago: None, - tx_packets: 0, - rx_packets: 0, - dropped: 0, - protocol_errors: 0, - path: "no data link".into(), - }); - } + // Collect all unique peer IDs known in this network. + let mut all_peer_ids: Vec = Vec::new(); + for member in &network.members { + if member.endpoint_id != status.endpoint_id + && !all_peer_ids.contains(&member.endpoint_id) + { + all_peer_ids.push(member.endpoint_id); } } - - // 2. Peers known to the pure WG plugin but not yet covered. if let Some(ref view) = pure_wg_view { for peer in &view.peers { - if peers - .iter() - .any(|p| p.endpoint_id == peer.endpoint_id.to_string()) - { - continue; + if !all_peer_ids.contains(&peer.endpoint_id) { + all_peer_ids.push(peer.endpoint_id); } - if let Some(tunnel) = &peer.tunnel { - peers.push(OverlayPeerReport { - endpoint_id: peer.endpoint_id.to_string(), - protocol: "wg".into(), - public_key: peer.public_key.to_string(), - address: peer.overlay_address_v4.map(|addr| addr.to_string()), - handshake_secs_ago: tunnel - .health - .since_handshake - .map(|s| s.as_secs()), - tx_packets: tunnel.stats.tx_packets, - rx_packets: tunnel.stats.rx_packets, - dropped: tunnel.stats.dropped_wrong_source - + tunnel.stats.dropped_oversize, - protocol_errors: tunnel.stats.protocol_errors, - path: tunnel.path.clone(), - }); - } else { - peers.push(OverlayPeerReport { - endpoint_id: peer.endpoint_id.to_string(), - protocol: "wg".into(), - public_key: peer.public_key.to_string(), - address: peer.overlay_address_v4.map(|addr| addr.to_string()), - handshake_secs_ago: None, - tx_packets: 0, - rx_packets: 0, - dropped: 0, - protocol_errors: 0, - path: "no data link".into(), - }); + } + } + if let Some(ref view) = wg_view { + for peer in &view.peers { + if !all_peer_ids.contains(&peer.endpoint_id) { + all_peer_ids.push(peer.endpoint_id); } } } - - // 3. tcp-tls peers not yet covered by either WG view. for report in &tcp_tls_reports { - if !peers + if !all_peer_ids.contains(&report.peer) { + all_peer_ids.push(report.peer); + } + } + + for peer_id in all_peer_ids { + let pure_wg_match = pure_wg_view + .as_ref() + .and_then(|v| v.peers.iter().find(|p| p.endpoint_id == peer_id)); + let tcp_tls_match = tcp_tls_reports.iter().find(|r| r.peer == peer_id); + let wg_quic_match = wg_view + .as_ref() + .and_then(|v| v.peers.iter().find(|p| p.endpoint_id == peer_id)); + + let member_addr = network + .members .iter() - .any(|p| p.endpoint_id == report.peer.to_string()) + .find(|m| m.endpoint_id == peer_id) + .and_then(|m| m.overlay_address_v4) + .map(|a| a.to_string()); + + if let Some(pure_peer) = pure_wg_match + && let Some(tunnel) = &pure_peer.tunnel { - let address = network - .members - .iter() - .find(|m| m.endpoint_id == report.peer) - .and_then(|m| m.overlay_address_v4) - .map(|a| a.to_string()); peers.push(OverlayPeerReport { - endpoint_id: report.peer.to_string(), - protocol: report.protocol.clone(), - public_key: String::new(), - address, - handshake_secs_ago: Some(report.uptime_secs), - tx_packets: report.tx_packets, - rx_packets: report.rx_packets, - dropped: report.dropped, + endpoint_id: peer_id.to_string(), + protocol: "wg".into(), + public_key: pure_peer.public_key.to_string(), + address: pure_peer + .overlay_address_v4 + .map(|a| a.to_string()) + .or(member_addr), + handshake_secs_ago: tunnel.health.since_handshake.map(|s| s.as_secs()), + tx_packets: tunnel.stats.tx_packets, + rx_packets: tunnel.stats.rx_packets, + dropped: tunnel.stats.dropped_wrong_source + + tunnel.stats.dropped_oversize, + protocol_errors: tunnel.stats.protocol_errors, + path: tunnel.path.clone(), + }); + } else if let Some(tcp_tls) = tcp_tls_match { + let pubkey = pure_wg_match + .map(|p| p.public_key.to_string()) + .or_else(|| wg_quic_match.map(|p| p.public_key.to_string())) + .unwrap_or_default(); + peers.push(OverlayPeerReport { + endpoint_id: peer_id.to_string(), + protocol: tcp_tls.protocol.clone(), + public_key: pubkey, + address: member_addr, + handshake_secs_ago: Some(tcp_tls.uptime_secs), + tx_packets: tcp_tls.tx_packets, + rx_packets: tcp_tls.rx_packets, + dropped: tcp_tls.dropped, protocol_errors: 0, - path: report.path.clone(), + path: tcp_tls.path.clone(), + }); + } else if let Some(wg_peer) = wg_quic_match + && let Some(tunnel) = &wg_peer.tunnel + { + peers.push(OverlayPeerReport { + endpoint_id: peer_id.to_string(), + protocol: "wg-quic".into(), + public_key: wg_peer.public_key.to_string(), + address: wg_peer + .overlay_address_v4 + .map(|a| a.to_string()) + .or(member_addr), + handshake_secs_ago: tunnel.health.since_handshake.map(|s| s.as_secs()), + tx_packets: tunnel.stats.tx_packets, + rx_packets: tunnel.stats.rx_packets, + dropped: tunnel.stats.dropped_wrong_source + + tunnel.stats.dropped_oversize, + protocol_errors: tunnel.stats.protocol_errors, + path: tunnel.path.clone(), + }); + } else if let Some(pure_peer) = pure_wg_match { + peers.push(OverlayPeerReport { + endpoint_id: peer_id.to_string(), + protocol: "wg".into(), + public_key: pure_peer.public_key.to_string(), + address: pure_peer + .overlay_address_v4 + .map(|a| a.to_string()) + .or(member_addr), + handshake_secs_ago: None, + tx_packets: 0, + rx_packets: 0, + dropped: 0, + protocol_errors: 0, + path: "no data link".into(), + }); + } else if let Some(wg_peer) = wg_quic_match { + peers.push(OverlayPeerReport { + endpoint_id: peer_id.to_string(), + protocol: "wg-quic".into(), + public_key: wg_peer.public_key.to_string(), + address: wg_peer + .overlay_address_v4 + .map(|a| a.to_string()) + .or(member_addr), + handshake_secs_ago: None, + tx_packets: 0, + rx_packets: 0, + dropped: 0, + protocol_errors: 0, + path: "no data link".into(), }); } } diff --git a/crates/tsunagi-wireguard/src/plugin.rs b/crates/tsunagi-wireguard/src/plugin.rs index f9c68be..ae2c7cb 100644 --- a/crates/tsunagi-wireguard/src/plugin.rs +++ b/crates/tsunagi-wireguard/src/plugin.rs @@ -51,12 +51,6 @@ pub fn local_ip_candidates() -> Vec { ips.push(IpAddr::V4(addr)); } } - for ip in iface.ipv6 { - let addr = ip.addr(); - if !addr.is_loopback() && !addr.is_unspecified() { - ips.push(IpAddr::V6(addr)); - } - } } } ips diff --git a/crates/tsunagi-wireguard/src/transport.rs b/crates/tsunagi-wireguard/src/transport.rs index ca9c66a..9ad0b64 100644 --- a/crates/tsunagi-wireguard/src/transport.rs +++ b/crates/tsunagi-wireguard/src/transport.rs @@ -3,7 +3,7 @@ //! Provides [`WireguardTransport`] implementing [`PacketTransport`] //! and [`WireguardLink`] implementing [`PacketLink`]. -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::net::SocketAddr; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex, RwLock}; @@ -31,6 +31,8 @@ pub struct WireguardLink { network: NetworkId, peer: EndpointId, target_addr: Arc>, + candidates: Arc>>, + confirmed: Arc, tx: mpsc::Sender, inbound_tx: mpsc::Sender, rx: tokio::sync::Mutex>, @@ -44,18 +46,23 @@ impl WireguardLink { network: NetworkId, peer: EndpointId, target_addr: SocketAddr, + candidates: Vec, socket: Arc, ) -> Self { let (tx_out, mut rx_out) = mpsc::channel::(1024); let (tx_in, rx_in) = mpsc::channel::(1024); let is_closed = Arc::new(AtomicBool::new(false)); + let confirmed = Arc::new(AtomicBool::new(false)); let closed_notify = Arc::new(Notify::new()); let target_holder = Arc::new(RwLock::new(target_addr)); + let candidates_holder = Arc::new(RwLock::new(candidates)); let socket_w = Arc::clone(&socket); let target_r = Arc::clone(&target_holder); + let candidates_r = Arc::clone(&candidates_holder); + let confirmed_r = Arc::clone(&confirmed); let is_closed_w = Arc::clone(&is_closed); let closed_notify_w = Arc::clone(&closed_notify); @@ -64,12 +71,30 @@ impl WireguardLink { if msg.len() > MAX_UDP_DATAGRAM_SIZE { continue; } - let target = match target_r.read() { - Ok(g) => *g, - Err(p) => *p.into_inner(), - }; - if let Err(err) = socket_w.send_to(&msg, target).await { - tracing::trace!(%err, %target, "failed to send WireGuard UDP datagram"); + if confirmed_r.load(Ordering::Acquire) { + let target = match target_r.read() { + Ok(g) => *g, + Err(p) => *p.into_inner(), + }; + if let Err(err) = socket_w.send_to(&msg, target).await { + tracing::trace!(%err, %target, "failed to send WireGuard UDP datagram"); + } + } else { + let targets = match candidates_r.read() { + Ok(g) => g.clone(), + Err(p) => p.into_inner().clone(), + }; + if targets.is_empty() { + let target = match target_r.read() { + Ok(g) => *g, + Err(p) => *p.into_inner(), + }; + let _ = socket_w.send_to(&msg, target).await; + } else { + for target in targets { + let _ = socket_w.send_to(&msg, target).await; + } + } } } is_closed_w.store(true, Ordering::Release); @@ -80,6 +105,8 @@ impl WireguardLink { network, peer, target_addr: target_holder, + candidates: candidates_holder, + confirmed, tx: tx_out, inbound_tx: tx_in, rx: tokio::sync::Mutex::new(rx_in), @@ -93,12 +120,24 @@ impl WireguardLink { if self.is_closed.load(Ordering::Acquire) { return false; } - // If queue is full, dropping is normal for an unreliable datagram link + self.confirmed.store(true, Ordering::Release); self.inbound_tx.try_send(payload).is_ok() } + /// Updates candidate addresses for this peer. + pub fn update_candidates(&self, addrs: impl IntoIterator) { + if let Ok(mut guard) = self.candidates.write() { + for addr in addrs { + if !guard.contains(&addr) { + guard.push(addr); + } + } + } + } + /// Updates the target socket address (roaming / NAT hole punch update). pub fn update_target(&self, new_target: SocketAddr) { + self.confirmed.store(true, Ordering::Release); if let Ok(mut guard) = self.target_addr.write() && *guard != new_target { @@ -177,7 +216,7 @@ pub struct WireguardTransport { listen_port: u16, socket: Arc, peer_addresses: RwLock>>, - peer_networks: RwLock>, + peer_networks: RwLock>>, links: RwLock>>, inbound_cb: Arc>>, } @@ -233,12 +272,15 @@ impl WireguardTransport { /// Associates a peer with a network ID. pub fn set_peer_network(&self, peer: EndpointId, network: NetworkId) { if let Ok(mut guard) = self.peer_networks.write() { - guard.insert(peer, network); + guard.entry(peer).or_default().insert(network); } } /// Records an observed UDP target address for a peer. pub fn set_peer_addr(&self, peer: EndpointId, addr: SocketAddr) { + if !addr.is_ipv4() { + return; + } if let Ok(mut guard) = self.peer_addresses.write() { let addrs = guard.entry(peer).or_default(); addrs.retain(|a| *a != addr); @@ -251,7 +293,7 @@ impl WireguardTransport { if let Ok(mut guard) = self.peer_addresses.write() { let entry = guard.entry(peer).or_default(); for addr in addrs { - if !entry.contains(&addr) { + if addr.is_ipv4() && !entry.contains(&addr) { entry.push(addr); } } @@ -278,12 +320,18 @@ impl WireguardTransport { addrs: &[SocketAddr], we_accept: bool, ) { - if let Ok(mut guard) = self.peer_networks.write() { - guard.insert(peer, network); - } - self.set_peer_addrs(peer, addrs.iter().copied()); + self.set_peer_network(peer, network); + let ipv4_addrs: Vec = addrs.iter().copied().filter(|a| a.is_ipv4()).collect(); + self.set_peer_addrs(peer, ipv4_addrs.iter().copied()); - if we_accept && !addrs.is_empty() { + // Update candidate addresses on any existing link for this peer + if let Ok(guard) = self.links.read() + && let Some(link) = guard.get(&(network, peer)) + { + link.update_candidates(ipv4_addrs.iter().copied()); + } + + if we_accept && !ipv4_addrs.is_empty() { let mut guard = match self.links.write() { Ok(g) => g, Err(p) => p.into_inner(), @@ -292,7 +340,8 @@ impl WireguardTransport { let link = Arc::new(WireguardLink::new( network, peer, - addrs[0], + ipv4_addrs[0], + ipv4_addrs, Arc::clone(&self.socket), )); entry.insert(Arc::clone(&link)); @@ -333,89 +382,109 @@ impl WireguardTransport { while let Ok((len, src_addr)) = socket.recv_from(&mut buf).await { let datagram = Bytes::copy_from_slice(&buf[..len]); - // 1. Check if an active link exists matching this peer / target - let matched_link = { + // 1. Identify which peer this packet comes from. + let matched_peer = { + let links_guard = match self.links.read() { + Ok(g) => g, + Err(p) => p.into_inner(), + }; + links_guard + .values() + .find(|link| { + link.target_addr() == src_addr || link.target_addr().ip() == src_addr.ip() + }) + .map(|link| link.peer()) + .or_else(|| { + let addrs_guard = match self.peer_addresses.read() { + Ok(g) => g, + Err(p) => p.into_inner(), + }; + addrs_guard + .iter() + .find(|(_, addrs)| { + addrs + .iter() + .any(|a| *a == src_addr || a.ip() == src_addr.ip()) + }) + .map(|(peer, _)| *peer) + }) + }; + + let Some(peer) = matched_peer else { + continue; + }; + + // 2. Deliver to ALL active links for this peer (across all networks this peer shares with us). + let peer_links: Vec> = { let guard = match self.links.read() { Ok(g) => g, Err(p) => p.into_inner(), }; guard .values() - .find(|link| { - link.target_addr() == src_addr || link.target_addr().ip() == src_addr.ip() - }) + .filter(|l| l.peer() == peer && !l.is_closed()) .cloned() + .collect() }; - if let Some(link) = matched_link { - link.update_target(src_addr); - link.deliver_inbound(datagram); + if !peer_links.is_empty() { + for link in peer_links { + link.update_target(src_addr); + link.deliver_inbound(datagram.clone()); + } continue; } - // 2. Check if a peer has candidate addresses matching src_addr - let matched_peer = { - let guard = match self.peer_addresses.read() { + // 3. If no active link exists for this peer yet, create inbound links for all networks known for this peer. + let networks: Vec = { + let guard = match self.peer_networks.read() { Ok(g) => g, Err(p) => p.into_inner(), }; guard - .iter() - .find(|(_, addrs)| { - addrs - .iter() - .any(|a| *a == src_addr || a.ip() == src_addr.ip()) - }) - .map(|(peer, _)| *peer) + .get(&peer) + .map(|s| s.iter().copied().collect()) + .unwrap_or_default() }; - if let Some(peer) = matched_peer { - let network = { - let guard = match self.peer_networks.read() { + for network in networks { + let link = { + let mut guard = match self.links.write() { Ok(g) => g, Err(p) => p.into_inner(), }; - guard.get(&peer).copied().or_else(|| { - if guard.len() == 1 { - guard.values().next().copied() - } else { - match self.links.read() { - Ok(g) => g.keys().next().map(|(net, _)| *net), - Err(p) => p.into_inner().keys().next().map(|(net, _)| *net), - } - } - }) - }; - - if let Some(network) = network { + if let Some(existing) = guard.get(&(network, peer)) + && !existing.is_closed() + { + existing.update_target(src_addr); + existing.deliver_inbound(datagram.clone()); + continue; + } + let candidates = self.peer_addrs(peer); let link = Arc::new(WireguardLink::new( network, peer, src_addr, + candidates, Arc::clone(&self.socket), )); - link.deliver_inbound(datagram); + link.update_target(src_addr); + link.deliver_inbound(datagram.clone()); + guard.insert((network, peer), Arc::clone(&link)); + link + }; - { - let mut guard = match self.links.write() { - Ok(g) => g, - Err(p) => p.into_inner(), - }; - guard.insert((network, peer), Arc::clone(&link)); - } - - let cb_guard = match self.inbound_cb.lock() { - Ok(g) => g, - Err(p) => p.into_inner(), - }; - if let Some(ref cb) = *cb_guard { - cb(InboundLink { - network, - peer, - protocol: crate::announcement::WG_PROTOCOL.to_string(), - link, - }); - } + let cb_guard = match self.inbound_cb.lock() { + Ok(g) => g, + Err(p) => p.into_inner(), + }; + if let Some(ref cb) = *cb_guard { + cb(InboundLink { + network, + peer, + protocol: crate::announcement::WG_PROTOCOL.to_string(), + link, + }); } } } @@ -454,12 +523,14 @@ impl PacketTransport for WireguardTransport { if let Some(existing) = guard.get(&(network, peer)) && !existing.is_closed() { + existing.update_candidates(addrs); return Ok(Arc::clone(existing) as SharedLink); } let link = Arc::new(WireguardLink::new( network, peer, target, + addrs, Arc::clone(&self.socket), )); guard.insert((network, peer), Arc::clone(&link)); diff --git a/crates/tsunagi-wireguard/tests/plugin_tests.rs b/crates/tsunagi-wireguard/tests/plugin_tests.rs index b9acdd6..0a5cbc5 100644 --- a/crates/tsunagi-wireguard/tests/plugin_tests.rs +++ b/crates/tsunagi-wireguard/tests/plugin_tests.rs @@ -204,3 +204,175 @@ async fn test_wireguard_plugin_end_to_end() { assert_eq!(received_a.1, peer_b); assert_eq!(received_a.2, packet_b_to_a); } + +#[tokio::test] +async fn test_wireguard_two_networks_same_peers() { + let identity_a = DeviceIdentity::generate(); + let peer_a = identity_a.endpoint_id(); + + let identity_b = DeviceIdentity::generate(); + let peer_b = identity_b.endpoint_id(); + + let net1 = NetworkId::from_bytes([11u8; 32]); + let net2 = NetworkId::from_bytes([22u8; 32]); + + let tmp_a = tempfile::tempdir().expect("tmp A"); + let tmp_b = tempfile::tempdir().expect("tmp B"); + + let transport_b = Arc::new( + WireguardTransport::bind(&identity_b, Some(0)) + .await + .expect("bind B"), + ); + let port_b = transport_b.bound_port(); + + let transport_a = Arc::new( + WireguardTransport::bind(&identity_a, Some(0)) + .await + .expect("bind A"), + ); + let port_a = transport_a.bound_port(); + + let plugin_a = + WireguardPlugin::open(WireguardConfig::new(tmp_a.path()), Arc::clone(&transport_a)) + .await + .expect("open plugin A"); + + let plugin_b = + WireguardPlugin::open(WireguardConfig::new(tmp_b.path()), Arc::clone(&transport_b)) + .await + .expect("open plugin B"); + + let (sink_a_tx, mut sink_a_rx) = mpsc::channel(20); + plugin_a.attach(PluginContext::with_sink(Arc::new(TestSink { + delivered: sink_a_tx, + }))); + + let (sink_b_tx, mut sink_b_rx) = mpsc::channel(20); + plugin_b.attach(PluginContext::with_sink(Arc::new(TestSink { + delivered: sink_b_tx, + }))); + + // Activate both networks + for net in [net1, net2] { + plugin_a.on_network_activated(net); + plugin_b.on_network_activated(net); + } + + // Exchange capabilities for both networks + for net in [net1, net2] { + let cap_a = loop { + if let Some(cap) = plugin_a.local_capability(net).expect("cap A") { + break cap; + } + tokio::time::sleep(Duration::from_millis(20)).await; + }; + let cap_b = loop { + if let Some(cap) = plugin_b.local_capability(net).expect("cap B") { + break cap; + } + tokio::time::sleep(Duration::from_millis(20)).await; + }; + plugin_a + .on_peer_capability(net, peer_b, &cap_b) + .expect("A on cap B"); + plugin_b + .on_peer_capability(net, peer_a, &cap_a) + .expect("B on cap A"); + } + + let addr_b: SocketAddr = format!("127.0.0.1:{port_b}").parse().unwrap(); + transport_a.set_peer_addr(peer_b, addr_b); + + let addr_a: SocketAddr = format!("127.0.0.1:{port_a}").parse().unwrap(); + transport_b.set_peer_addr(peer_a, addr_a); + + // Accept loops + let transport_a_clone = Arc::clone(&transport_a); + tokio::spawn(async move { + transport_a_clone.accept_loop(|_| {}).await; + }); + + let (inbound_tx, mut inbound_rx) = mpsc::channel::(10); + let transport_b_clone = Arc::clone(&transport_b); + tokio::spawn(async move { + transport_b_clone + .accept_loop(move |inbound| { + let _ = inbound_tx.try_send(inbound); + }) + .await; + }); + + // Open links for both networks from A to B + for net in [net1, net2] { + let link_a = transport_a + .open(net, peer_b, "wg") + .await + .expect("open A link"); + plugin_a.on_peer_link(net, peer_b, link_a); + } + + // Deliver inbound links to B + for _ in 0..2 { + let inbound = tokio::time::timeout(Duration::from_secs(5), inbound_rx.recv()) + .await + .expect("inbound timeout") + .expect("inbound option"); + plugin_b.on_peer_link(inbound.network, peer_a, inbound.link); + } + + // Wait for both handshakes to complete on both networks + for net in [net1, net2] { + let mut up = false; + for _ in 0..50 { + if let Some(overview) = plugin_a.overview(net) + && overview.peers.iter().any(|p| p.is_up()) + { + up = true; + break; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + assert!(up, "handshake on {net:?} should complete"); + } + + // Send on net1 + let pkt1 = sample_ipv4_packet(); + assert!(plugin_a.carry(net1, peer_b, pkt1.clone())); + let recv1 = tokio::time::timeout(Duration::from_secs(5), sink_b_rx.recv()) + .await + .expect("recv1 timeout") + .expect("recv1 option"); + assert_eq!(recv1.0, net1); + assert_eq!(recv1.2, pkt1); + + // Send on net2 from A to B + let mut pkt2_bytes = sample_ipv4_packet().to_vec(); + pkt2_bytes[15] = 99; // different byte + let pkt2 = Bytes::from(pkt2_bytes); + assert!(plugin_a.carry(net2, peer_b, pkt2.clone())); + let recv2 = tokio::time::timeout(Duration::from_secs(5), sink_b_rx.recv()) + .await + .expect("recv2 timeout") + .expect("recv2 option"); + assert_eq!(recv2.0, net2); + assert_eq!(recv2.2, pkt2); + + // Send reply on net1 from B to A + assert!(plugin_b.carry(net1, peer_a, pkt1.clone())); + let reply1 = tokio::time::timeout(Duration::from_secs(5), sink_a_rx.recv()) + .await + .expect("reply1 timeout") + .expect("reply1 option"); + assert_eq!(reply1.0, net1); + assert_eq!(reply1.2, pkt1); + + // Send reply on net2 from B to A + assert!(plugin_b.carry(net2, peer_a, pkt2.clone())); + let reply2 = tokio::time::timeout(Duration::from_secs(5), sink_a_rx.recv()) + .await + .expect("reply2 timeout") + .expect("reply2 option"); + assert_eq!(reply2.0, net2); + assert_eq!(reply2.2, pkt2); +}