From 35acc4d231b001ddb6d4cdea85412d1eb2f053e9 Mon Sep 17 00:00:00 2001 From: Aleksandr Bogomiakov Date: Thu, 1 Oct 2026 02:38:13 +0100 Subject: [PATCH] Fix WireGuard relay preemption with UDP probing, allow VM bridges, and categorize status addresses When remote peers communicate across networks with mixed connectivity (e.g. a remote Linux node, a local Linux host, and a Windows VM guest under KVM): - Previously, `WireguardTransport` immediately created and installed unverified direct links into `RelayHub` in `notify_peer_capability` and `open()`. When a remote peer had only unroutable private addresses (e.g. 192.168.122.151 on a KVM NAT bridge), this dead direct link preempted the working 2-hop relay, causing packet drops, failed handshakes, and REKEY_TIMEOUT errors. - Host-guest VM bridges (`virbr*`) were previously filtered out, preventing the VM guest from discovering the host's bridge IP (192.168.122.1) and forcing it to attempt cross-interface connections to host Wi-Fi/LAN. - Concurrent activation of multiple networks sharing one `tsun0` interface raced in `sync_addresses`, resulting in netlink `File exists (os error 17)` and `The object already exists` on Windows. Key fixes: - In `crates/tsunagi-wireguard/src/transport.rs`: - Implement bidirectional UDP probing with `TSUNAGI_WG_PROBE_V1` PING/PONG. - In `open()`, probe candidate endpoints with retransmission up to 1200ms. If unreachable, return `TransportError::Unreachable`, preventing dead links from preempting valid relay routes in `RelayHub`. - In `notify_peer_capability`, send hole-punching probes without prematurely installing unverified links. - In `accept_loop`, reply to PING probes with PONG and install confirmed inbound links targeting the verified source address. - In `WireguardLink::send`, inspect byte 74 of enveloped packets to detect handshake initiations, ensuring confirmed links route data packets only to the active target. - In `crates/tsunagi-wireguard/src/plugin.rs`: - Permit VM bridges (`virbr*`) as candidate interfaces (priority 10). - Add `prioritize_local_subnet` to prioritize candidates sharing a local IPv4 subnet with the receiver (e.g. 192.168.122.0/24). - In `crates/tsunagi/src/overlay/interface.rs` & provisioners: - Add async mutex `sync_lock` to serialize concurrent address syncs across networks. - Handle netlink EEXIST (os error 17) and Windows already-exists errors idempotently as success. - In `crates/tsunagi-cli/src/main.rs`: - Split `local addresses` in `tsunagi status` into `announced` (candidate addresses in priority order with interface and rank), `not announced` (container bridges/internal addresses with interface name), and `loopback`. - Tests: - Add `test_wireguard_transport_unreachable_probe_times_out` and `test_matches_subnet` unit tests. 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 | 90 ++++- crates/tsunagi-tcp-tls/src/lib.rs | 4 +- crates/tsunagi-wireguard/src/plugin.rs | 63 ++- crates/tsunagi-wireguard/src/transport.rs | 366 ++++++++++++++---- .../tests/transport_tests.rs | 34 ++ crates/tsunagi/src/overlay/interface.rs | 3 + crates/tsunagi/src/overlay/provision/linux.rs | 15 +- .../tsunagi/src/overlay/provision/windows.rs | 7 +- 8 files changed, 479 insertions(+), 103 deletions(-) diff --git a/crates/tsunagi-cli/src/main.rs b/crates/tsunagi-cli/src/main.rs index 2046cc4..d45ccef 100644 --- a/crates/tsunagi-cli/src/main.rs +++ b/crates/tsunagi-cli/src/main.rs @@ -3015,23 +3015,73 @@ async fn addresses_section() -> report::Section { use report::{Health, Row, Section}; let mut addresses = Section::new("local addresses"); - let found = netwatch_addresses().await; - if found.is_empty() { + let found = netwatch_interface_addresses().await; + + let mut candidates = tsunagi_wireguard::local_ip_candidates(); + for ip in tsunagi_tcp_tls::local_ip_candidates() { + if !candidates.contains(&ip) { + candidates.push(ip); + } + } + + if found.is_empty() && candidates.is_empty() { addresses.push( Row::new(Health::Degraded, "interfaces", "none found") .with_note("best effort; the agent may still find a way out"), ); + return addresses; } - for addr in found { - // Loopback alone reaches nobody, but on a host that also has a real - // address it is unremarkable, so it is labelled rather than flagged. - let kind = match (addr.is_loopback(), addr.is_ipv4()) { - (true, _) => "loopback", - (false, true) => "ipv4", - (false, false) => "ipv6", - }; - addresses.push(Row::new(Health::Info, kind, addr.to_string())); + + let mut iface_map: std::collections::HashMap = + std::collections::HashMap::new(); + for item in &found { + iface_map.insert(item.addr, item.iface.clone()); } + + // 1. Announced peer candidate addresses (in candidate priority order) + let mut seen = std::collections::HashSet::new(); + for (idx, &cand) in candidates.iter().enumerate() { + if seen.insert(cand) { + let note = iface_map + .get(&cand) + .map(|iface| format!("{iface} · candidate #{}", idx + 1)); + let row = Row::new(Health::Info, "announced", cand.to_string()); + addresses.push(match note { + Some(n) => row.with_note(n), + None => row, + }); + } + } + + // 2. Unannounced local addresses + let mut unannounced = Vec::new(); + let mut loopbacks = Vec::new(); + + for item in &found { + if seen.contains(&item.addr) { + continue; + } + if item.addr.is_loopback() { + loopbacks.push(item); + } else { + unannounced.push(item); + } + } + + unannounced.sort_by_key(|item| item.addr); + for item in unannounced { + let note = format!("{} · not announced to peers", item.iface); + addresses + .push(Row::new(Health::Info, "not announced", item.addr.to_string()).with_note(note)); + } + + // 3. Loopback addresses + loopbacks.sort_by_key(|item| item.addr); + for item in loopbacks { + addresses + .push(Row::new(Health::Info, "loopback", item.addr.to_string()).with_note(&item.iface)); + } + addresses } @@ -3433,21 +3483,31 @@ fn program_path() -> String { .unwrap_or_else(|| "tsunagi".to_string()) } -async fn netwatch_addresses() -> Vec { +struct LocalAddressInfo { + addr: std::net::IpAddr, + iface: String, +} + +async fn netwatch_interface_addresses() -> Vec { // Best effort; used for diagnostics only. let state = netwatch::interfaces::State::new().await; let mut addresses = Vec::new(); + let mut seen = std::collections::HashSet::new(); for (name, iface) in &state.interfaces { let name_lower = name.to_lowercase(); if name_lower.starts_with("tsun") || name_lower.contains("tsunagi") { continue; } for prefix in iface.addrs() { - addresses.push(prefix.addr()); + let addr = prefix.addr(); + if seen.insert(addr) { + addresses.push(LocalAddressInfo { + addr, + iface: name.clone(), + }); + } } } - addresses.sort(); - addresses.dedup(); addresses } diff --git a/crates/tsunagi-tcp-tls/src/lib.rs b/crates/tsunagi-tcp-tls/src/lib.rs index 7a2a992..c72bec9 100644 --- a/crates/tsunagi-tcp-tls/src/lib.rs +++ b/crates/tsunagi-tcp-tls/src/lib.rs @@ -5,5 +5,7 @@ pub mod crypto; pub mod plugin; pub mod transport; -pub use plugin::{TCP_TLS_PROTOCOL, TCP_TLS_VERSION, TcpTlsCodec, TcpTlsConfig, TcpTlsPlugin}; +pub use plugin::{ + TCP_TLS_PROTOCOL, TCP_TLS_VERSION, TcpTlsCodec, TcpTlsConfig, TcpTlsPlugin, local_ip_candidates, +}; pub use transport::TcpTlsTransport; diff --git a/crates/tsunagi-wireguard/src/plugin.rs b/crates/tsunagi-wireguard/src/plugin.rs index c77db7d..ccc20df 100644 --- a/crates/tsunagi-wireguard/src/plugin.rs +++ b/crates/tsunagi-wireguard/src/plugin.rs @@ -93,11 +93,10 @@ pub(crate) fn is_candidate_interface(iface: &netdev::Interface) -> bool { return false; } - // Skip container, VM, and virtual bridge interfaces + // Skip container interfaces and virtual bridges purely internal to container engines if name.starts_with("docker") || name.starts_with("br-") || name.starts_with("veth") - || name.starts_with("virbr") || name.starts_with("cni") || name.starts_with("flannel") || name.starts_with("cali") @@ -105,12 +104,7 @@ pub(crate) fn is_candidate_interface(iface: &netdev::Interface) -> bool { || name.starts_with("dummy") || name.starts_with("lxc") || name.starts_with("podman") - || friendly.contains("vethernet") - || friendly.contains("vboxnet") - || friendly.contains("vmnet") - || desc.contains("virtualbox") - || desc.contains("hyper-v") - || desc.contains("vmware") + || friendly.contains("wsl") { return false; } @@ -118,6 +112,43 @@ pub(crate) fn is_candidate_interface(iface: &netdev::Interface) -> bool { true } +/// Reorders candidate addresses so that candidates on the same local subnet as one of our +/// active network interfaces appear first. +pub fn prioritize_local_subnet(candidates: &mut [SocketAddr]) { + let local_subnets: Vec<(Ipv4Addr, u8)> = netdev::get_interfaces() + .into_iter() + .filter(|iface| iface.is_up() && !iface.is_loopback()) + .flat_map(|iface| { + iface + .ipv4 + .into_iter() + .map(|net| (net.addr(), net.prefix_len())) + }) + .collect(); + + candidates.sort_by_key(|addr| { + let is_local = match addr.ip() { + IpAddr::V4(v4) => local_subnets + .iter() + .any(|(local_ip, prefix_len)| matches_subnet(*local_ip, *prefix_len, v4)), + IpAddr::V6(_) => false, + }; + if is_local { 0 } else { 1 } + }); +} + +fn matches_subnet(local_ip: Ipv4Addr, prefix_len: u8, target_ip: Ipv4Addr) -> bool { + if prefix_len == 0 { + return false; + } + let mask = if prefix_len >= 32 { + !0u32 + } else { + !0u32 << (32 - prefix_len) + }; + (u32::from(local_ip) & mask) == (u32::from(target_ip) & mask) +} + pub(crate) fn is_candidate_ipv4(addr: std::net::Ipv4Addr) -> bool { if addr.is_loopback() || addr.is_unspecified() || addr.is_link_local() || addr.is_broadcast() { return false; @@ -925,15 +956,15 @@ mod tests { iface.name = "veth12345".to_string(); assert!(!is_candidate_interface(&iface)); - // Libvirt NAT bridge virbr* - must be filtered + // Libvirt NAT bridge virbr* - allowed for VM guests/hosts iface.name = "virbr0".to_string(); - assert!(!is_candidate_interface(&iface)); + assert!(is_candidate_interface(&iface)); // CNI bridge - must be filtered iface.name = "cni0".to_string(); assert!(!is_candidate_interface(&iface)); - // Hyper-V / WSL vEthernet switch on Windows - must be filtered + // WSL vEthernet switch on Windows - must be filtered iface.name = "{A1B2C3D4-E5F6-7890-1234-567890ABCDEF}".to_string(); iface.friendly_name = Some("vEthernet (WSL)".to_string()); iface.description = Some("Hyper-V Virtual Ethernet Adapter".to_string()); @@ -998,4 +1029,14 @@ mod tests { // LAN IP with gateway should score higher than VPN IP without gateway assert!(p_lan > p_ts); } + + #[test] + fn test_matches_subnet() { + let local_ip = Ipv4Addr::new(192, 168, 122, 151); + let same_subnet = Ipv4Addr::new(192, 168, 122, 1); + let diff_subnet = Ipv4Addr::new(192, 168, 1, 2); + + assert!(matches_subnet(local_ip, 24, same_subnet)); + assert!(!matches_subnet(local_ip, 24, diff_subnet)); + } } diff --git a/crates/tsunagi-wireguard/src/transport.rs b/crates/tsunagi-wireguard/src/transport.rs index e675c7d..e49ee02 100644 --- a/crates/tsunagi-wireguard/src/transport.rs +++ b/crates/tsunagi-wireguard/src/transport.rs @@ -11,7 +11,7 @@ use std::sync::{Arc, Mutex, RwLock}; use bytes::Bytes; use iroh::EndpointId; use tokio::net::UdpSocket; -use tokio::sync::{Notify, mpsc}; +use tokio::sync::{Notify, mpsc, oneshot}; use tsunagi::BoxFuture; use tsunagi::dataplane::transport::{ @@ -25,6 +25,57 @@ pub const DEFAULT_PORT: u16 = 51820; /// Maximum payload length for a single framed datagram over UDP. pub const MAX_UDP_DATAGRAM_SIZE: usize = 65507; +const PROBE_MAGIC: &[u8; 19] = b"TSUNAGI_WG_PROBE_V1"; +const PROBE_TYPE_PING: u8 = 1; +const PROBE_TYPE_PONG: u8 = 2; +const PROBE_LEN: usize = 19 + 1 + 32 + 32 + 32 + 8; // 124 bytes + +fn encode_probe( + msg_type: u8, + sender: EndpointId, + receiver: EndpointId, + network: NetworkId, + cookie: u64, +) -> Vec { + let mut buf = Vec::with_capacity(PROBE_LEN); + buf.extend_from_slice(PROBE_MAGIC); + buf.push(msg_type); + buf.extend_from_slice(sender.as_bytes()); + buf.extend_from_slice(receiver.as_bytes()); + buf.extend_from_slice(network.as_bytes()); + buf.extend_from_slice(&cookie.to_be_bytes()); + buf +} + +struct DecodedProbe { + msg_type: u8, + sender: EndpointId, + receiver: EndpointId, + network: NetworkId, + cookie: u64, +} + +fn decode_probe(bytes: &[u8]) -> Option { + if bytes.len() < PROBE_LEN || &bytes[..19] != PROBE_MAGIC { + return None; + } + let msg_type = bytes[19]; + if msg_type != PROBE_TYPE_PING && msg_type != PROBE_TYPE_PONG { + return None; + } + let sender = EndpointId::from_bytes(&bytes[20..52].try_into().ok()?).ok()?; + let receiver = EndpointId::from_bytes(&bytes[52..84].try_into().ok()?).ok()?; + let network = NetworkId::from_bytes(bytes[84..116].try_into().ok()?); + let cookie = u64::from_be_bytes(bytes[116..124].try_into().ok()?); + Some(DecodedProbe { + msg_type, + sender, + receiver, + network, + cookie, + }) +} + /// An authenticated datagram link carried over direct UDP. #[derive(Debug)] pub struct WireguardLink { @@ -71,7 +122,16 @@ impl WireguardLink { if msg.len() > MAX_UDP_DATAGRAM_SIZE { continue; } - let is_handshake_init = msg.first() == Some(&1); + // Check if this is a WireGuard handshake initiation packet: + // An enveloped packet has a 74-byte header (envelope::HEADER). + // A raw packet may directly have WireGuard type at index 0. + let is_handshake_init = if msg.len() >= 74 + 4 && msg[0] == 1 { + msg[74] == 1 + } else if !msg.is_empty() { + msg[0] == 1 + } else { + false + }; if confirmed_r.load(Ordering::Acquire) && !is_handshake_init { let target = match target_r.read() { Ok(g) => *g, @@ -125,6 +185,16 @@ impl WireguardLink { self.inbound_tx.try_send(payload).is_ok() } + /// Marks this link as confirmed by a probe response or verified packet. + pub fn confirm(&self) { + self.confirmed.store(true, Ordering::Release); + } + + /// Whether this link has had verified two-way direct connectivity. + pub fn is_confirmed(&self) -> bool { + self.confirmed.load(Ordering::Acquire) + } + /// Updates candidate addresses for this peer. pub fn update_candidates(&self, addrs: impl IntoIterator) { if let Ok(mut guard) = self.candidates.write() { @@ -215,6 +285,7 @@ impl PacketLink for WireguardLink { } type InboundCallback = Box; +type PendingProbes = Arc>>>; /// Direct UDP WireGuard packet transport. pub struct WireguardTransport { @@ -225,6 +296,7 @@ pub struct WireguardTransport { peer_networks: RwLock>>, links: RwLock>>, inbound_cb: Arc>>, + pending_probes: PendingProbes, } impl std::fmt::Debug for WireguardTransport { @@ -267,6 +339,7 @@ impl WireguardTransport { peer_networks: RwLock::new(HashMap::new()), links: RwLock::new(HashMap::new()), inbound_cb: Arc::new(Mutex::new(None)), + pending_probes: Arc::new(Mutex::new(HashMap::new())), }) } @@ -317,8 +390,9 @@ impl WireguardTransport { /// Notifies the transport that a peer's capability was announced. /// - /// When `we_accept` is true (i.e. `local_id > peer`), the transport can proactively - /// create and install an inbound link so both ends are ready and can punch NAT. + /// When `we_accept` is true, sends a hole-punch probe to candidate addresses + /// to open stateful NAT firewalls, but does NOT create an unverified direct link + /// in the hub before packets actually flow. pub fn notify_peer_capability( &self, network: NetworkId, @@ -337,35 +411,17 @@ impl WireguardTransport { link.update_candidates(ipv4_addrs.iter().copied()); } + // Send hole punching probe if we accept and have candidate addresses. if we_accept && !ipv4_addrs.is_empty() { - let mut guard = match self.links.write() { - Ok(g) => g, - Err(p) => p.into_inner(), - }; - if let std::collections::hash_map::Entry::Vacant(entry) = guard.entry((network, peer)) { - let link = Arc::new(WireguardLink::new( - network, - peer, - ipv4_addrs[0], - ipv4_addrs, - Arc::clone(&self.socket), - )); - entry.insert(Arc::clone(&link)); - drop(guard); - - 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 socket = Arc::clone(&self.socket); + let local_id = self.local_id; + tokio::spawn(async move { + let cookie: u64 = rand::random(); + let probe = encode_probe(PROBE_TYPE_PING, local_id, peer, network, cookie); + for addr in ipv4_addrs { + let _ = socket.send_to(&probe, addr).await; } - } + }); } } @@ -389,49 +445,156 @@ impl WireguardTransport { if src_addr.ip().is_unspecified() { continue; } - let datagram = Bytes::copy_from_slice(&buf[..len]); - // 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) - .map(|link| link.peer()) - .or_else(|| { - let addrs_guard = match self.peer_addresses.read() { + // 1. Direct UDP probe packet handling + if len >= PROBE_LEN + && &buf[..19] == PROBE_MAGIC + && let Some(probe) = decode_probe(&buf[..len]) + { + if probe.receiver != self.local_id { + continue; + } + + if probe.msg_type == PROBE_TYPE_PING { + let is_known = { + let guard = match self.peer_networks.read() { Ok(g) => g, Err(p) => p.into_inner(), }; - addrs_guard - .iter() - .find(|(_, addrs)| addrs.contains(&src_addr)) - .map(|(peer, _)| *peer) - .or_else(|| { - links_guard - .values() - .find(|link| link.target_addr().ip() == src_addr.ip()) - .map(|link| link.peer()) - .or_else(|| { - addrs_guard - .iter() - .find(|(_, addrs)| { - addrs.iter().any(|a| a.ip() == src_addr.ip()) - }) - .map(|(peer, _)| *peer) - }) - }) - }) + guard + .get(&probe.sender) + .is_some_and(|nets| nets.contains(&probe.network)) + }; + if is_known { + let pong = encode_probe( + PROBE_TYPE_PONG, + self.local_id, + probe.sender, + probe.network, + probe.cookie, + ); + let _ = socket.send_to(&pong, src_addr).await; + + self.set_peer_addr(probe.sender, src_addr); + + let link = { + let mut guard = match self.links.write() { + Ok(g) => g, + Err(p) => p.into_inner(), + }; + if let Some(existing) = guard.get(&(probe.network, probe.sender)) + && !existing.is_closed() + { + existing.update_target(src_addr); + existing.confirm(); + None + } else { + let candidates = self.peer_addrs(probe.sender); + let link = Arc::new(WireguardLink::new( + probe.network, + probe.sender, + src_addr, + candidates, + Arc::clone(&self.socket), + )); + link.confirm(); + guard.insert((probe.network, probe.sender), Arc::clone(&link)); + Some(link) + } + }; + + if let Some(link) = 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: probe.network, + peer: probe.sender, + protocol: crate::announcement::WG_PROTOCOL.to_string(), + link, + }); + } + } + } + continue; + } else if probe.msg_type == PROBE_TYPE_PONG { + self.set_peer_addr(probe.sender, src_addr); + let waiter = { + let mut probes = match self.pending_probes.lock() { + Ok(g) => g, + Err(p) => p.into_inner(), + }; + probes.remove(&(probe.network, probe.sender, probe.cookie)) + }; + if let Some(tx) = waiter { + let _ = tx.send(src_addr); + } + continue; + } + } + + // 2. Data / handshake packet handling + let datagram = Bytes::copy_from_slice(&buf[..len]); + + // Identify which peer this packet comes from + let matched_peer = { + // If it's an envelope packet (VERSION 1, len >= 74), read source endpoint ID directly + let from_envelope = if len >= 74 && buf[0] == 1 { + let dest = &buf[34..66]; + if dest == self.local_id.as_bytes() { + EndpointId::from_bytes(&buf[2..34].try_into().unwrap_or([0u8; 32])).ok() + } else { + None + } + } else { + None + }; + + from_envelope.or_else(|| { + 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) + .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.contains(&src_addr)) + .map(|(peer, _)| *peer) + .or_else(|| { + links_guard + .values() + .find(|link| link.target_addr().ip() == src_addr.ip()) + .map(|link| link.peer()) + .or_else(|| { + addrs_guard + .iter() + .find(|(_, addrs)| { + addrs.iter().any(|a| 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). + self.set_peer_addr(peer, src_addr); + + // 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, @@ -452,7 +615,7 @@ impl WireguardTransport { continue; } - // 3. If no active link exists for this peer yet, create inbound links for all networks known for this peer. + // 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, @@ -524,13 +687,75 @@ impl PacketTransport for WireguardTransport { _protocol: &'a str, ) -> BoxFuture<'a, Result> { Box::pin(async move { - let addrs = self.peer_addrs(peer); - let Some(target) = addrs.first().copied() else { + let mut addrs = self.peer_addrs(peer); + if addrs.is_empty() { return Err(TransportError::Unreachable(format!( "no candidate UDP address known for peer {}", peer.fmt_short() ))); - }; + } + crate::plugin::prioritize_local_subnet(&mut addrs); + + // If an existing confirmed link is alive, reuse it + { + let guard = match self.links.read() { + Ok(g) => g, + Err(p) => p.into_inner(), + }; + if let Some(existing) = guard.get(&(network, peer)) + && !existing.is_closed() + && existing.is_confirmed() + { + existing.update_candidates(addrs); + return Ok(Arc::clone(existing) as SharedLink); + } + } + + // Probe candidate addresses with PING + let cookie: u64 = rand::random(); + let (tx, rx) = oneshot::channel::(); + { + let mut probes = match self.pending_probes.lock() { + Ok(g) => g, + Err(p) => p.into_inner(), + }; + probes.insert((network, peer, cookie), tx); + } + + let probe_ping = encode_probe(PROBE_TYPE_PING, self.local_id, peer, network, cookie); + let socket = Arc::clone(&self.socket); + let candidates = addrs.clone(); + + let probe_task = tokio::spawn(async move { + for i in 0..3 { + for target in &candidates { + let _ = socket.send_to(&probe_ping, *target).await; + } + if i < 2 { + tokio::time::sleep(std::time::Duration::from_millis(350)).await; + } + } + }); + + let confirmed_addr = + match tokio::time::timeout(std::time::Duration::from_millis(1200), rx).await { + Ok(Ok(addr)) => { + probe_task.abort(); + addr + } + _ => { + probe_task.abort(); + let mut probes = match self.pending_probes.lock() { + Ok(g) => g, + Err(p) => p.into_inner(), + }; + probes.remove(&(network, peer, cookie)); + return Err(TransportError::Unreachable(format!( + "direct UDP probe to peer {} timed out", + peer.fmt_short() + ))); + } + }; let link = { let mut guard = match self.links.write() { @@ -540,16 +765,19 @@ impl PacketTransport for WireguardTransport { if let Some(existing) = guard.get(&(network, peer)) && !existing.is_closed() { + existing.update_target(confirmed_addr); existing.update_candidates(addrs); + existing.confirm(); return Ok(Arc::clone(existing) as SharedLink); } let link = Arc::new(WireguardLink::new( network, peer, - target, + confirmed_addr, addrs, Arc::clone(&self.socket), )); + link.confirm(); guard.insert((network, peer), Arc::clone(&link)); link }; diff --git a/crates/tsunagi-wireguard/tests/transport_tests.rs b/crates/tsunagi-wireguard/tests/transport_tests.rs index 9cf1618..0d1c1fc 100644 --- a/crates/tsunagi-wireguard/tests/transport_tests.rs +++ b/crates/tsunagi-wireguard/tests/transport_tests.rs @@ -110,3 +110,37 @@ async fn test_wireguard_transport_loopback() { assert!(link_a.path_description().contains("wireguard via")); assert!(link_b.path_description().contains("wireguard via")); } + +#[tokio::test] +async fn test_wireguard_transport_unreachable_probe_times_out() { + let identity_a = DeviceIdentity::generate(); + let identity_b = DeviceIdentity::generate(); + let peer_b = identity_b.endpoint_id(); + let network = NetworkId::from_bytes([99u8; 32]); + + let transport_a = Arc::new( + WireguardTransport::bind(&identity_a, Some(0)) + .await + .expect("bind client should succeed"), + ); + + // Register a non-responsive address for peer B (TEST-NET-1 documentation IP which drops packets) + let unreachable_addr: SocketAddr = "192.0.2.1:51820".parse().unwrap(); + transport_a.set_peer_addr(peer_b, unreachable_addr); + transport_a.set_peer_network(peer_b, network); + + // Spawn accept loop on A + let transport_a_clone = Arc::clone(&transport_a); + tokio::spawn(async move { + transport_a_clone.accept_loop(|_| {}).await; + }); + + // Opening to unreachable candidate must fail with Unreachable rather than returning a dead direct link + let result = transport_a.open(network, peer_b, "wg").await; + match result { + Err(tsunagi::dataplane::transport::TransportError::Unreachable(msg)) => { + assert!(msg.contains("timed out")); + } + other => panic!("expected Unreachable error, got {other:?}"), + } +} diff --git a/crates/tsunagi/src/overlay/interface.rs b/crates/tsunagi/src/overlay/interface.rs index 5b8ce02..3ae5f49 100644 --- a/crates/tsunagi/src/overlay/interface.rs +++ b/crates/tsunagi/src/overlay/interface.rs @@ -152,6 +152,7 @@ pub struct Interface { /// re-applied on every pass. applied: std::sync::Mutex>, task: std::sync::Mutex>>, + sync_lock: Arc>, } impl Interface { @@ -228,6 +229,7 @@ impl Interface { tally, applied: std::sync::Mutex::new(Vec::new()), task: std::sync::Mutex::new(Some(task)), + sync_lock: Arc::new(tokio::sync::Mutex::new(())), }) } @@ -237,6 +239,7 @@ impl Interface { /// The interface is never recreated for it: that would drop every tunnel /// riding on it for the sake of one address. pub async fn sync_addresses(&self) -> Result<(), OverlayError> { + let _guard = self.sync_lock.lock().await; let wanted = self.wanted_addresses(); { let applied = match self.applied.lock() { diff --git a/crates/tsunagi/src/overlay/provision/linux.rs b/crates/tsunagi/src/overlay/provision/linux.rs index d8b5497..b50623c 100644 --- a/crates/tsunagi/src/overlay/provision/linux.rs +++ b/crates/tsunagi/src/overlay/provision/linux.rs @@ -520,16 +520,19 @@ async fn configure_link(configure: &Configure) -> Result<(), OverlayError> { } for cidr in &configure.add { - handle + let res = handle .address() .add(index, cidr.addr, cidr.prefix_len) .execute() - .await - .map_err(|err| { - OverlayError::Unavailable(format!( + .await; + if let Err(err) = res { + let err_str = err.to_string(); + if !err_str.contains("File exists") && !err_str.contains("os error 17") { + return Err(OverlayError::Unavailable(format!( "cannot add {cidr} to interface `{name}`: {err}" - )) - })?; + ))); + } + } } Ok(()) diff --git a/crates/tsunagi/src/overlay/provision/windows.rs b/crates/tsunagi/src/overlay/provision/windows.rs index 28769e2..37f9965 100644 --- a/crates/tsunagi/src/overlay/provision/windows.rs +++ b/crates/tsunagi/src/overlay/provision/windows.rs @@ -262,7 +262,12 @@ async fn configure(plan: &InterfacePlan, changes: &super::Changes) -> Result<(), set_mtu(&plan.name, mtu).await?; } for cidr in &changes.add { - address_add(&plan.name, *cidr).await?; + if let Err(err) = address_add(&plan.name, *cidr).await { + let err_str = err.to_string(); + if !err_str.contains("already exists") && !err_str.contains("0x80071392") { + return Err(err); + } + } } if changes.bring_up { // Best effort: a Wintun adapter with a session is up already, and a