Prevent WireGuard timer loop failure spam and fix handshake broadcast
When a WireGuard tunnel handshake timed out or a rekey failed,
Inner::on_tunnel_failed closed the transport link and reported the failure
to the agent core. However:
1. The peer remained in WireguardDevice::inner.peers. Once boringtun
declared ConnectionExpired (after 90s), self.tunn.is_expired() remained
permanently true on all subsequent 250ms timer ticks. drive_timers
repeatedly detected failed = true every 250ms, flooding logs with
'WARN WireGuard tunnel failed due to rekey timeout, triggering fallback'
and sending repeated protocol failure events to the core 4 times/sec.
2. In WireguardLink::send, is_handshake_init incorrectly expected an
envelope header offset (msg[74] == 1) when checking for initiation
packets, which failed on raw WireGuard UDP packets. As a result,
handshake initiations were not broadcast to all candidate endpoints.
3. In WireguardTransport::accept_loop, packet source matching contained
dead envelope destination matching inspecting ephemeral key bytes
against local endpoint IDs.
Fixes:
- In crates/tsunagi-wireguard/src/device.rs:
- Add failed: bool to FlowTunnel to latch failure state. Once a failure
is reported, update_timers returns (TunnResult::Done, false) on all
subsequent ticks until a new handshake is started or completed.
- In Inner::on_tunnel_failed, remove the failed peer from inner.peers so
drive_timers ceases ticking on the dead tunnel.
- In crates/tsunagi-wireguard/src/plugin.rs:
- In Worker::sync, retain only active unclosed links in state.links.
- In crates/tsunagi-wireguard/src/transport.rs:
- Check is_handshake_init directly as '!msg.is_empty() && msg[0] == 1',
ensuring handshake initiations are broadcast to all candidate addresses.
- Clean up matched_peer to look up links and peer candidate addresses directly.
- Tests:
- Update test_flow_tunnel_rekey_timeout_triggers_failure to assert failure
latching and verify no repeated failure reports on subsequent ticks.
Validation on Linux:
- cargo fmt --all -- --check passed.
- cargo clippy --locked --workspace --all-targets -- -D warnings passed.
- cargo test --locked --workspace --all-targets passed.
- cargo build --release passed.
This commit is contained in:
Aleksandr Bogomiakov committed
1 parent
8cfece9479
commit
d2778f0b5a
3 files changed
+59
-57
No files matched your search
@@ -36,6 +36,7 @@ struct FlowTunnel {
|
||||
queued: VecDeque<FlowId>,
|
||||
handshake_started_at: Option<std::time::Instant>,
|
||||
rekey_timeouts: u32,
|
||||
failed: bool,
|
||||
}
|
||||
|
||||
fn is_data(result: &TunnResult<'_>) -> bool {
|
||||
@@ -58,6 +59,7 @@ impl FlowTunnel {
|
||||
{
|
||||
self.handshake_started_at = Some(std::time::Instant::now());
|
||||
self.rekey_timeouts = 0;
|
||||
self.failed = false;
|
||||
}
|
||||
if self.queued.len() < 256 {
|
||||
self.queued.push_back(flow);
|
||||
@@ -74,6 +76,7 @@ impl FlowTunnel {
|
||||
if self.tunn.time_since_last_handshake().is_some() {
|
||||
self.handshake_started_at = None;
|
||||
self.rekey_timeouts = 0;
|
||||
self.failed = false;
|
||||
}
|
||||
let mut flow = 0;
|
||||
if packet.is_empty() {
|
||||
@@ -87,12 +90,17 @@ impl FlowTunnel {
|
||||
}
|
||||
|
||||
fn update_timers<'a>(&mut self, scratch: &'a mut [u8]) -> (TunnResult<'a>, bool) {
|
||||
if self.failed {
|
||||
return (TunnResult::Done, false);
|
||||
}
|
||||
|
||||
let result = self.tunn.update_timers(scratch);
|
||||
if matches!(
|
||||
result,
|
||||
TunnResult::Err(boringtun::noise::errors::WireGuardError::ConnectionExpired)
|
||||
) || self.tunn.is_expired()
|
||||
{
|
||||
self.failed = true;
|
||||
self.queued.clear();
|
||||
self.handshake_started_at = None;
|
||||
self.rekey_timeouts = 0;
|
||||
@@ -110,6 +118,7 @@ impl FlowTunnel {
|
||||
if let Some(started) = self.handshake_started_at
|
||||
&& started.elapsed() >= HANDSHAKE_TIMEOUT
|
||||
{
|
||||
self.failed = true;
|
||||
self.queued.clear();
|
||||
self.handshake_started_at = None;
|
||||
self.rekey_timeouts = 0;
|
||||
@@ -251,8 +260,9 @@ struct Inner {
|
||||
}
|
||||
|
||||
impl Inner {
|
||||
fn on_tunnel_failed(&self, peer: EndpointId) {
|
||||
fn on_tunnel_failed(&self, peer: EndpointId, public_key: WgPublicKey) {
|
||||
self.transport.close_link(self.network, peer);
|
||||
write_lock(&self.peers).remove(&public_key);
|
||||
self.context.report_peer_protocol_failure(
|
||||
self.network,
|
||||
peer,
|
||||
@@ -376,6 +386,7 @@ impl WireguardDevice {
|
||||
queued: VecDeque::new(),
|
||||
handshake_started_at: None,
|
||||
rekey_timeouts: 0,
|
||||
failed: false,
|
||||
}),
|
||||
link,
|
||||
counters: Arc::new(PeerCounters::default()),
|
||||
@@ -472,6 +483,7 @@ fn kick_handshake(peer: &Peer) {
|
||||
TunnResult::WriteToNetwork(out) => {
|
||||
tunn.handshake_started_at = Some(std::time::Instant::now());
|
||||
tunn.rekey_timeouts = 0;
|
||||
tunn.failed = false;
|
||||
Some(out.len())
|
||||
}
|
||||
_ => None,
|
||||
@@ -597,7 +609,7 @@ async fn drive_timers(inner: Arc<Inner>) {
|
||||
peer = %peer.endpoint_id.fmt_short(),
|
||||
"WireGuard tunnel failed due to rekey timeout, triggering fallback"
|
||||
);
|
||||
inner.on_tunnel_failed(peer.endpoint_id);
|
||||
inner.on_tunnel_failed(peer.endpoint_id, peer.public_key);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -626,6 +638,7 @@ mod tests {
|
||||
queued: VecDeque::new(),
|
||||
handshake_started_at: Some(std::time::Instant::now() - Duration::from_secs(16)),
|
||||
rekey_timeouts: 0,
|
||||
failed: false,
|
||||
};
|
||||
|
||||
let mut scratch = vec![0u8; SCRATCH];
|
||||
@@ -633,6 +646,14 @@ mod tests {
|
||||
let (_, failed) = tunnel.update_timers(&mut scratch);
|
||||
assert!(failed, "handshake timeout should report failure");
|
||||
assert!(tunnel.handshake_started_at.is_none());
|
||||
assert!(tunnel.failed, "tunnel should be marked as failed");
|
||||
|
||||
// Subsequent timer tick must not report failure again
|
||||
let (_, failed_again) = tunnel.update_timers(&mut scratch);
|
||||
assert!(
|
||||
!failed_again,
|
||||
"subsequent timer tick should not re-trigger failure"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
@@ -654,6 +675,7 @@ mod tests {
|
||||
queued: VecDeque::new(),
|
||||
handshake_started_at: Some(std::time::Instant::now()),
|
||||
rekey_timeouts: 2,
|
||||
failed: true,
|
||||
};
|
||||
|
||||
let mut scratch = vec![0u8; SCRATCH];
|
||||
|
||||
@@ -605,6 +605,8 @@ impl Worker {
|
||||
return;
|
||||
};
|
||||
|
||||
state.links.retain(|_, link| !link.is_closed());
|
||||
|
||||
let allocations = state.allocations.clone();
|
||||
let mut wanted: Vec<WgPublicKey> = Vec::new();
|
||||
for (endpoint_id, announcement) in &state.announcements {
|
||||
|
||||
@@ -122,16 +122,8 @@ impl WireguardLink {
|
||||
if msg.len() > MAX_UDP_DATAGRAM_SIZE {
|
||||
continue;
|
||||
}
|
||||
// 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
|
||||
};
|
||||
// In WireGuard protocol, type 1 is Handshake Initiation (148 bytes).
|
||||
let is_handshake_init = !msg.is_empty() && msg[0] == 1;
|
||||
if confirmed_r.load(Ordering::Acquire) && !is_handshake_init {
|
||||
let target = match target_r.read() {
|
||||
Ok(g) => *g,
|
||||
@@ -570,52 +562,38 @@ impl WireguardTransport {
|
||||
|
||||
// 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
|
||||
let links_guard = match self.links.read() {
|
||||
Ok(g) => g,
|
||||
Err(p) => p.into_inner(),
|
||||
};
|
||||
|
||||
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)
|
||||
})
|
||||
})
|
||||
})
|
||||
})
|
||||
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 {
|
||||
|
||||
Reference in new issue
Block a user