From b93c3bfed612453f3e620fb96f9c8a6f808cb323 Mon Sep 17 00:00:00 2001 From: Wade Simmons Date: Thu, 1 Oct 2026 09:15:40 -0500 Subject: [PATCH] multiport: fix review findings in lane probing and recv_error - Don't probe lanes on a tunnel idle in both directions, and don't count lane probes or their acks as tunnel traffic. A lane's own keepalive was keeping idle tunnels from ever being dropped by tunnels.drop_inactive. - Accept a recv_error from an up lane's target. The peer answers lane traffic from the lane socket it arrived on, and the spoof guard only accepted the base remote, so recv_errors for lane traffic were ignored. - Keep the probe generation across a lane reset, so an ack for a probe sent before a roam can't promote the lane on the new path. - Bring handshake.proto in line with payload.go: field 3 is reserved and TxLanes is field 4. --- connection_manager.go | 22 ++++-------- design/multiport.md | 14 +++++--- handshake/handshake.proto | 12 ++++--- inside.go | 6 ++-- lanes.go | 34 +++++++++++++++--- lanes_test.go | 74 +++++++++++++++++++++++++++++++++++++++ outside.go | 13 ++++--- 7 files changed, 138 insertions(+), 37 deletions(-) diff --git a/connection_manager.go b/connection_manager.go index 06bc6402..cf0ad99c 100644 --- a/connection_manager.go +++ b/connection_manager.go @@ -197,26 +197,18 @@ func (cm *connectionManager) doTrafficCheck(localIndex uint32, p, nb, out []byte } cm.resetRelayTrafficCheck(hostinfo) - cm.maintainLanes(localIndex, decision, hostinfo, now, nb, out) + cm.maintainLanes(decision, hostinfo, now, nb, out) } // maintainLanes piggybacks multiport lane probing on the per-tunnel traffic -// tick. This tick is the right place for it precisely because lanes are -// demand-driven: a tunnel only lands here when it has traffic, which is the -// same condition that raises lane demand. -// -// makeTrafficDecision returns a nil hostinfo on some keep-alive paths, so -// re-resolve the index in that case. -func (cm *connectionManager) maintainLanes(localIndex uint32, decision trafficDecision, hostinfo *HostInfo, now time.Time, nb, out []byte) { - if decision == deleteTunnel || decision == closeTunnel { +// tick. makeTrafficDecision returns a nil hostinfo for a tunnel that is idle in +// both directions, and that tunnel is skipped: probing it would cost a probe per +// lane per keepalive for a tunnel nothing is using. Its lanes are probed again +// on the first tick after traffic resumes. +func (cm *connectionManager) maintainLanes(decision trafficDecision, hostinfo *HostInfo, now time.Time, nb, out []byte) { + if hostinfo == nil || decision == deleteTunnel || decision == closeTunnel { return } - if hostinfo == nil { - hostinfo = cm.hostMap.QueryIndex(localIndex) - if hostinfo == nil { - return - } - } cm.intf.probeLanes(hostinfo, now, nb, out) } diff --git a/design/multiport.md b/design/multiport.md index f65a7af4..8c49e7fe 100644 --- a/design/multiport.md +++ b/design/multiport.md @@ -203,8 +203,11 @@ everything it needs in one atomic read and there is no window where one is set and the other is not. Probing is driven by the connection manager's per-tunnel traffic tick -(`maintainLanes` -> `probeLanes`), which only fires for a tunnel with traffic -- -the same condition that makes a lane worth having. Timers: +(`maintainLanes` -> `probeLanes`), which skips a tunnel that is idle in both +directions -- a lane is only worth having while there is traffic to put on it. +Probes and their acks are not counted as tunnel traffic, so a lane's own +keepalive cannot keep an otherwise idle tunnel from being dropped by +`tunnels.drop_inactive`. Timers: | timer | value | |---|---| @@ -445,9 +448,10 @@ and lane probe *acks* all use the base session and a socket on the base port would see the tunnel's address move and roam-thrash. `recv_error` is the one deliberate exception: it replies from the socket the -offending packet arrived on, because a lane peer's spoof guard compares our -source address against that lane's remote and would discard a reply from the -base port. +offending packet arrived on, so the reply to lane traffic travels the reverse of +the lane's own 4-tuple, the only one the peer's NAT and stateful firewalls are +known to pass. The receiving side's spoof guard accepts a `recv_error` from the +base remote or from the target of any of its up lanes (`isLaneRemote`). ### Wire compatibility diff --git a/handshake/handshake.proto b/handshake/handshake.proto index 352aebd7..accaabf0 100644 --- a/handshake/handshake.proto +++ b/handshake/handshake.proto @@ -32,12 +32,14 @@ message NebulaHandshakeDetails { LaneDetails ResponderLanes = 10; } -// LaneDetails advertises a host's multiport lane capability. On a base -// handshake LaneIndex is 0 and PortCount/BasePort describe the sender's -// consecutively bound UDP ports. On a lane handshake the initiator sets -// LaneIndex to its (nonzero) lane number. +// LaneDetails advertises a host's multiport lane capability. PortCount/BasePort +// describe the sender's consecutively bound UDP ports, and TxLanes is how many +// lanes the sender may send on, so the peer knows how many lane sessions it must +// be prepared to receive on. message LaneDetails { uint32 PortCount = 1; uint32 BasePort = 2; - uint32 LaneIndex = 3; + // was a per-lane handshake index + reserved 3; + uint32 TxLanes = 4; } diff --git a/inside.go b/inside.go index 05980aad..a77a4313 100644 --- a/inside.go +++ b/inside.go @@ -611,9 +611,9 @@ func (f *Interface) sendNoMetrics(t header.MessageType, st header.MessageSubType //l.WithField("trace", string(debug.Stack())).Error("out Header ", &Header{Version, t, st, 0, hostinfo.remoteIndexId, c}, p) out = header.Encode(out, header.Version, t, st, hostinfo.remoteIndexId, c) // A closing tunnel is torn down right after this, so skip the connection manager entirely: no point recording - // traffic or asking the lighthouse for a punch. Otherwise, if we rebound since this tunnel last sent, ask the - // lighthouse to get the far side punching at us again. - if t != header.CloseTunnel && f.connectionManager.Out(hostinfo) { + // traffic or asking the lighthouse for a punch. Lane probe acks are not traffic either. Otherwise, if we rebound + // since this tunnel last sent, ask the lighthouse to get the far side punching at us again. + if t != header.CloseTunnel && !isLaneProbe(t, st) && f.connectionManager.Out(hostinfo) { f.lightHouse.QueryServer(hostinfo.vpnAddrs[0]) if f.l.Enabled(context.Background(), slog.LevelDebug) { f.l.Debug("Lighthouse update triggered for punch due to rebind epoch", diff --git a/lanes.go b/lanes.go index dcdd009e..19cb56bd 100644 --- a/lanes.go +++ b/lanes.go @@ -236,7 +236,7 @@ func newLaneSet(r *handshake.Result, myLanes int, myAddr, peerAddr netip.Addr) * // and the kernel remembers that for as long as the flow stays busy. A flow that // starts while the lanes are still down therefore gets pinned to queue 0 on // both hosts for its whole life. This costs one probe per lane on any tunnel - // with traffic; an idle tunnel is never ticked, so it still costs nothing. + // with traffic; an idle tunnel is never probed, so it still costs nothing. for s := 1; s < txLanes; s++ { ls.demand[s].Store(true) } @@ -545,8 +545,8 @@ func (ls *laneSet) noteAck(s int, gen uint8, now time.Time) (netip.AddrPort, boo // probeLanes runs one lane maintenance pass for a peer: it demotes lanes whose // probe went unanswered, re-proves lanes that have been up a while without one, // and probes down lanes the data plane asked for. Driven by the connection -// manager's per-tunnel traffic tick, which only fires for a live tunnel — the -// same condition that produces lane demand in the first place. +// manager's per-tunnel traffic tick, which skips a tunnel that is idle in both +// directions. func (f *Interface) probeLanes(hostinfo *HostInfo, now time.Time, nb, out []byte) { ls := hostinfo.lanes if ls == nil || ls.txLanes < 2 { @@ -618,6 +618,10 @@ func (f *Interface) probeLanes(hostinfo *HostInfo, now time.Time, nb, out []byte // resetLocked takes every lane down and clears its probe state, without // counting it as a failure. // +// The probe generation survives the reset. Starting it over would let a late ack +// for a probe sent before the reset match the first probe sent after it, and +// promote the lane to a target nothing has proven. +// // Demand is deliberately left standing: it records that a routine has real // traffic for this peer, which a roam or a relay detour does not change. Keeping // it re-probes the lanes that were actually carrying data as soon as a path @@ -625,10 +629,32 @@ func (f *Interface) probeLanes(hostinfo *HostInfo, now time.Time, nb, out []byte func (ls *laneSet) resetLocked() { for s := 1; s < len(ls.probe); s++ { ls.txAddr[s].Store(nil) - ls.probe[s] = laneProbeState{} + ls.probe[s] = laneProbeState{gen: ls.probe[s].gen} } } +// isLaneProbe reports whether t/st is a lane probe or its ack. Probing is upkeep +// the lanes do for themselves, not traffic: counting it would let a lane's own +// keepalive keep an idle tunnel from ever being dropped. +func isLaneProbe(t header.MessageType, st header.MessageSubType) bool { + return t == header.Test && (st == header.LaneProbe || st == header.LaneProbeAck) +} + +// isLaneRemote reports whether addr is the target of one of our up lanes. The +// peer answers a lane packet from the socket it arrived on, so a recv_error for +// lane traffic comes from that lane's target rather than the base remote. +func (ls *laneSet) isLaneRemote(addr netip.AddrPort) bool { + if ls == nil { + return false + } + for s := 1; s < ls.txLanes; s++ { + if a := ls.txAddr[s].Load(); a != nil && *a == addr { + return true + } + } + return false +} + // sendLaneProbe sends a probe on lane s to addr from a socket on lane s's port. // The probe is an ordinary Test packet encrypted with the lane's session, so an // ack proves the whole lane: our source port reached the peer, its reply reached diff --git a/lanes_test.go b/lanes_test.go index 39745be2..cdf39418 100644 --- a/lanes_test.go +++ b/lanes_test.go @@ -593,6 +593,14 @@ func TestLaneProbeRoamResets(t *testing.T) { assert.Equal(t, netip.MustParseAddr("198.51.100.7"), ls.peerAddr) ls.mu.Unlock() + // A late ack for the probe sent before the roam must not promote the lane + // on the new path, so the generation has to carry across the reset. + ls.demand[1].Store(true) + ifce.probeLanes(hi, now, nb, out) + _, promoted = ls.noteAck(1, gen, now) + assert.False(t, promoted, "an ack from before the roam promoted the lane") + assert.Nil(t, ls.txAddr[1].Load()) + // Relayed (no direct remote) means no lanes at all. ls.demand[1].Store(true) ifce.probeLanes(hi, now, nb, out) @@ -606,6 +614,71 @@ func TestLaneProbeRoamResets(t *testing.T) { assert.Nil(t, ls.txAddr[1].Load(), "lane survived losing the direct path") } +// A peer answers lane traffic it can't place with a recv_error from the lane +// socket it arrived on, so the spoof guard must accept one from an up lane's +// target and still refuse one from anywhere else. +func TestHandleRecvErrorFromLane(t *testing.T) { + hostMap := newHostMap(test.NewLogger()) + ifce := newLaneTestInterface(hostMap) + + initR, _ := runTestHandshake(t) + ls := newTestLaneSet(t, initR, 4, 4, 4242, 4) + hi := newTestLaneHostInfo(t, initR, ls) + hostMap.unlockedAddHostInfo(hi, ifce) + + now := time.Now() + ifce.probeLanes(hi, now, make([]byte, 12), make([]byte, mtu)) + ls.mu.Lock() + gen := ls.probe[1].gen + ls.mu.Unlock() + laneAddr, promoted := ls.noteAck(1, gen, now) + require.True(t, promoted) + require.NotEqual(t, hi.GetRemote(), laneAddr) + + h := &header.H{Type: header.RecvError, RemoteIndex: hi.remoteIndexId} + + ifce.handleRecvError(netip.MustParseAddrPort("192.0.2.1:9999"), h) + require.NotNil(t, hostMap.QueryReverseIndex(hi.remoteIndexId), "a recv_error from an unknown port closed the tunnel") + + ifce.handleRecvError(laneAddr, h) + assert.Nil(t, hostMap.QueryReverseIndex(hi.remoteIndexId), "a recv_error from an up lane's target was ignored") +} + +// An idle tunnel is not probed, so its lanes' keepalives can't keep it alive. +// Probing resumes on the first tick that sees traffic. +func TestIdleTunnelSkipsLaneProbes(t *testing.T) { + hostMap := newHostMap(test.NewLogger()) + ifce := newLaneTestInterface(hostMap) + + initR, _ := runTestHandshake(t) + ls := newTestLaneSet(t, initR, 4, 4, 4242, 4) + hi := newTestLaneHostInfo(t, initR, ls) + hostMap.unlockedAddHostInfo(hi, ifce) + + // An empty pool fails every cert, which with pki.disconnect_invalid off + // leaves the tunnel up. + ifce.pki.caPool.Store(cert.NewCAPool()) + + cm := ifce.connectionManager + p := []byte("") + nb := make([]byte, 12) + out := make([]byte, mtu) + probed := func() bool { + ls.mu.Lock() + defer ls.mu.Unlock() + return !ls.probe[1].sentAt.IsZero() + } + + // Adding the tunnel records outbound traffic; drop it so the tick sees idle. + hi.takeTraffic() + cm.doTrafficCheck(hi.localIndexId, p, nb, out, time.Now()) + assert.False(t, probed(), "an idle tunnel was probed") + + cm.In(hi) + cm.doTrafficCheck(hi.localIndexId, p, nb, out, time.Now()) + assert.True(t, probed(), "a tunnel with traffic was not probed") +} + // A lane probe is answered on the base tunnel, echoing the header's lane, so a // peer cannot get us to vouch for a lane it never probed. func TestHandleLaneProbe(t *testing.T) { @@ -629,6 +702,7 @@ func TestHandleLaneProbe(t *testing.T) { assert.Equal(t, header.Test, h.Type) assert.Equal(t, header.LaneProbeAck, h.Subtype) assert.Equal(t, uint8(0), h.Lane(), "the ack is base-tunnel traffic") + assert.False(t, hi.sentSinceCheck(), "a probe ack must not count as tunnel traffic") pt, err := laneSessionFor(t, newTestLaneSet(t, initR, 4, 4, 4242, 4), 1).dKey.DecryptDanger( nil, sent.bufs[0][:header.Len], sent.bufs[0][header.Len:], h.MessageCounter, make([]byte, 12)) diff --git a/outside.go b/outside.go index 199337fe..2faa6db5 100644 --- a/outside.go +++ b/outside.go @@ -186,7 +186,9 @@ func (f *Interface) readOutsidePackets(via ViaSender, packet []byte, rxc *rxCont if lane == 0 { f.handleHostRoaming(hostinfo, via) } - f.connectionManager.In(hostinfo) + if !isLaneProbe(h.Type, h.Subtype) { + f.connectionManager.In(hostinfo) + } switch h.Type { case header.Message: @@ -552,9 +554,10 @@ func (f *Interface) maybeSendRecvError(endpoint netip.AddrPort, index uint32, q } } -// sendRecvError replies from the socket the offending packet arrived on (q). -// A lane peer's spoof guard compares our source addr against the lane's -// remote, so a reply from the base port would be discarded. +// sendRecvError replies from the socket the offending packet arrived on (q), so +// the reply to lane traffic rides the reverse of the lane's own 4-tuple: the +// only one the peer's NAT and stateful firewalls are known to let through. The +// peer's handleRecvError accepts it from that lane's target. func (f *Interface) sendRecvError(endpoint netip.AddrPort, index uint32, q int) { f.messageMetrics.Tx(header.RecvError, 0, 1) @@ -591,7 +594,7 @@ func (f *Interface) handleRecvError(addr netip.AddrPort, h *header.H) { } hr := hostinfo.GetRemote() - if hr.IsValid() && hr != addr { + if hr.IsValid() && hr != addr && !hostinfo.lanes.isLaneRemote(addr) { f.l.Info("Someone spoofing recv_errors?", "addr", addr, "hostinfoRemote", hr,