ab/tsunagi
1
0 Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Fix WireGuard relay preemption with UDP probing, allow VM bridges, and categorize status addresses

Browse Source 
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).
This commit is contained in:
Aleksandr Bogomiakov committed 2026-10-01 02:38:13 +01:00
1 parent c791102ad9
commit 35acc4d231
8 files changed
+479 -103
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

crates/tsunagi-cli/src/main.rs
+75 -15
View File
@@ -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::net::IpAddr, String> =
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<std::net::IpAddr> {
struct LocalAddressInfo {
addr: std::net::IpAddr,
iface: String,
}
async fn netwatch_interface_addresses() -> Vec<LocalAddressInfo> {
// 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
}
crates/tsunagi-tcp-tls/src/lib.rs
+3 -1
View File
@@ -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;
crates/tsunagi-wireguard/src/plugin.rs
+52 -11
View File
@@ -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));
}
}
crates/tsunagi-wireguard/src/transport.rs
+297 -69
View File
@@ -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<u8> {
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<DecodedProbe> {
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<Item = SocketAddr>) {
if let Ok(mut guard) = self.candidates.write() {
@@ -215,6 +285,7 @@ impl PacketLink for WireguardLink {
}
type InboundCallback = Box<dyn Fn(InboundLink) + Send + Sync + 'static>;
type PendingProbes = Arc<Mutex<HashMap<(NetworkId, EndpointId, u64), oneshot::Sender<SocketAddr>>>>;
/// Direct UDP WireGuard packet transport.
pub struct WireguardTransport {
@@ -225,6 +296,7 @@ pub struct WireguardTransport {
peer_networks: RwLock<HashMap<EndpointId, HashSet<NetworkId>>>,
links: RwLock<HashMap<(NetworkId, EndpointId), Arc<WireguardLink>>>,
inbound_cb: Arc<Mutex<Option<InboundCallback>>>,
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<Arc<WireguardLink>> = {
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<NetworkId> = {
let guard = match self.peer_networks.read() {
Ok(g) => g,
@@ -524,13 +687,75 @@ impl PacketTransport for WireguardTransport {
_protocol: &'a str,
) -> BoxFuture<'a, Result<SharedLink, TransportError>> {
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::<SocketAddr>();
{
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
};
crates/tsunagi-wireguard/tests/transport_tests.rs
+34
View File
@@ -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:?}"),
}
}
crates/tsunagi/src/overlay/interface.rs
+3
View File
@@ -152,6 +152,7 @@ pub struct Interface {
/// re-applied on every pass.
applied: std::sync::Mutex<Vec<Cidr>>,
task: std::sync::Mutex<Option<JoinHandle<()>>>,
sync_lock: Arc<tokio::sync::Mutex<()>>,
}
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() {
crates/tsunagi/src/overlay/provision/linux.rs
+9 -6
View File
@@ -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(())
crates/tsunagi/src/overlay/provision/windows.rs
+6 -1
View File
@@ -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