mirror of
https://github.com/slackhq/nebula.git
synced 2026-11-15 20:56:52 +01:00
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.
This commit is contained in:
1 parent
4782c5da9c
commit
b93c3bfed6
7 files changed
+138
-37
No files matched your search
+7
-15
@@ -197,26 +197,18 @@ func (cm *connectionManager) doTrafficCheck(localIndex uint32, p, nb, out []byte
|
|||||||
}
|
}
|
||||||
|
|
||||||
cm.resetRelayTrafficCheck(hostinfo)
|
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
|
// maintainLanes piggybacks multiport lane probing on the per-tunnel traffic
|
||||||
// tick. This tick is the right place for it precisely because lanes are
|
// tick. makeTrafficDecision returns a nil hostinfo for a tunnel that is idle in
|
||||||
// demand-driven: a tunnel only lands here when it has traffic, which is the
|
// both directions, and that tunnel is skipped: probing it would cost a probe per
|
||||||
// same condition that raises lane demand.
|
// lane per keepalive for a tunnel nothing is using. Its lanes are probed again
|
||||||
//
|
// on the first tick after traffic resumes.
|
||||||
// makeTrafficDecision returns a nil hostinfo on some keep-alive paths, so
|
func (cm *connectionManager) maintainLanes(decision trafficDecision, hostinfo *HostInfo, now time.Time, nb, out []byte) {
|
||||||
// re-resolve the index in that case.
|
if hostinfo == nil || decision == deleteTunnel || decision == closeTunnel {
|
||||||
func (cm *connectionManager) maintainLanes(localIndex uint32, decision trafficDecision, hostinfo *HostInfo, now time.Time, nb, out []byte) {
|
|
||||||
if decision == deleteTunnel || decision == closeTunnel {
|
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
if hostinfo == nil {
|
|
||||||
hostinfo = cm.hostMap.QueryIndex(localIndex)
|
|
||||||
if hostinfo == nil {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
}
|
|
||||||
cm.intf.probeLanes(hostinfo, now, nb, out)
|
cm.intf.probeLanes(hostinfo, now, nb, out)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
+9
-5
@@ -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.
|
and the other is not.
|
||||||
|
|
||||||
Probing is driven by the connection manager's per-tunnel traffic tick
|
Probing is driven by the connection manager's per-tunnel traffic tick
|
||||||
(`maintainLanes` -> `probeLanes`), which only fires for a tunnel with traffic --
|
(`maintainLanes` -> `probeLanes`), which skips a tunnel that is idle in both
|
||||||
the same condition that makes a lane worth having. Timers:
|
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 |
|
| 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.
|
would see the tunnel's address move and roam-thrash.
|
||||||
|
|
||||||
`recv_error` is the one deliberate exception: it replies from the socket the
|
`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
|
offending packet arrived on, so the reply to lane traffic travels the reverse of
|
||||||
source address against that lane's remote and would discard a reply from the
|
the lane's own 4-tuple, the only one the peer's NAT and stateful firewalls are
|
||||||
base port.
|
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
|
### Wire compatibility
|
||||||
|
|
||||||
|
|||||||
@@ -32,12 +32,14 @@ message NebulaHandshakeDetails {
|
|||||||
LaneDetails ResponderLanes = 10;
|
LaneDetails ResponderLanes = 10;
|
||||||
}
|
}
|
||||||
|
|
||||||
// LaneDetails advertises a host's multiport lane capability. On a base
|
// LaneDetails advertises a host's multiport lane capability. PortCount/BasePort
|
||||||
// handshake LaneIndex is 0 and PortCount/BasePort describe the sender's
|
// describe the sender's consecutively bound UDP ports, and TxLanes is how many
|
||||||
// consecutively bound UDP ports. On a lane handshake the initiator sets
|
// lanes the sender may send on, so the peer knows how many lane sessions it must
|
||||||
// LaneIndex to its (nonzero) lane number.
|
// be prepared to receive on.
|
||||||
message LaneDetails {
|
message LaneDetails {
|
||||||
uint32 PortCount = 1;
|
uint32 PortCount = 1;
|
||||||
uint32 BasePort = 2;
|
uint32 BasePort = 2;
|
||||||
uint32 LaneIndex = 3;
|
// was a per-lane handshake index
|
||||||
|
reserved 3;
|
||||||
|
uint32 TxLanes = 4;
|
||||||
}
|
}
|
||||||
@@ -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)
|
//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)
|
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
|
// 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
|
// traffic or asking the lighthouse for a punch. Lane probe acks are not traffic either. Otherwise, if we rebound
|
||||||
// lighthouse to get the far side punching at us again.
|
// 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) {
|
if t != header.CloseTunnel && !isLaneProbe(t, st) && f.connectionManager.Out(hostinfo) {
|
||||||
f.lightHouse.QueryServer(hostinfo.vpnAddrs[0])
|
f.lightHouse.QueryServer(hostinfo.vpnAddrs[0])
|
||||||
if f.l.Enabled(context.Background(), slog.LevelDebug) {
|
if f.l.Enabled(context.Background(), slog.LevelDebug) {
|
||||||
f.l.Debug("Lighthouse update triggered for punch due to rebind epoch",
|
f.l.Debug("Lighthouse update triggered for punch due to rebind epoch",
|
||||||
|
|||||||
@@ -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
|
// 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
|
// 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
|
// 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++ {
|
for s := 1; s < txLanes; s++ {
|
||||||
ls.demand[s].Store(true)
|
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
|
// 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,
|
// 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
|
// 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
|
// manager's per-tunnel traffic tick, which skips a tunnel that is idle in both
|
||||||
// same condition that produces lane demand in the first place.
|
// directions.
|
||||||
func (f *Interface) probeLanes(hostinfo *HostInfo, now time.Time, nb, out []byte) {
|
func (f *Interface) probeLanes(hostinfo *HostInfo, now time.Time, nb, out []byte) {
|
||||||
ls := hostinfo.lanes
|
ls := hostinfo.lanes
|
||||||
if ls == nil || ls.txLanes < 2 {
|
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
|
// resetLocked takes every lane down and clears its probe state, without
|
||||||
// counting it as a failure.
|
// 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
|
// 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
|
// 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
|
// 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() {
|
func (ls *laneSet) resetLocked() {
|
||||||
for s := 1; s < len(ls.probe); s++ {
|
for s := 1; s < len(ls.probe); s++ {
|
||||||
ls.txAddr[s].Store(nil)
|
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.
|
// 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
|
// 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
|
// ack proves the whole lane: our source port reached the peer, its reply reached
|
||||||
|
|||||||
@@ -593,6 +593,14 @@ func TestLaneProbeRoamResets(t *testing.T) {
|
|||||||
assert.Equal(t, netip.MustParseAddr("198.51.100.7"), ls.peerAddr)
|
assert.Equal(t, netip.MustParseAddr("198.51.100.7"), ls.peerAddr)
|
||||||
ls.mu.Unlock()
|
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.
|
// Relayed (no direct remote) means no lanes at all.
|
||||||
ls.demand[1].Store(true)
|
ls.demand[1].Store(true)
|
||||||
ifce.probeLanes(hi, now, nb, out)
|
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")
|
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
|
// 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.
|
// peer cannot get us to vouch for a lane it never probed.
|
||||||
func TestHandleLaneProbe(t *testing.T) {
|
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.Test, h.Type)
|
||||||
assert.Equal(t, header.LaneProbeAck, h.Subtype)
|
assert.Equal(t, header.LaneProbeAck, h.Subtype)
|
||||||
assert.Equal(t, uint8(0), h.Lane(), "the ack is base-tunnel traffic")
|
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(
|
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))
|
nil, sent.bufs[0][:header.Len], sent.bufs[0][header.Len:], h.MessageCounter, make([]byte, 12))
|
||||||
|
|||||||
+8
-5
@@ -186,7 +186,9 @@ func (f *Interface) readOutsidePackets(via ViaSender, packet []byte, rxc *rxCont
|
|||||||
if lane == 0 {
|
if lane == 0 {
|
||||||
f.handleHostRoaming(hostinfo, via)
|
f.handleHostRoaming(hostinfo, via)
|
||||||
}
|
}
|
||||||
f.connectionManager.In(hostinfo)
|
if !isLaneProbe(h.Type, h.Subtype) {
|
||||||
|
f.connectionManager.In(hostinfo)
|
||||||
|
}
|
||||||
|
|
||||||
switch h.Type {
|
switch h.Type {
|
||||||
case header.Message:
|
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).
|
// sendRecvError replies from the socket the offending packet arrived on (q), so
|
||||||
// A lane peer's spoof guard compares our source addr against the lane's
|
// the reply to lane traffic rides the reverse of the lane's own 4-tuple: the
|
||||||
// remote, so a reply from the base port would be discarded.
|
// 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) {
|
func (f *Interface) sendRecvError(endpoint netip.AddrPort, index uint32, q int) {
|
||||||
f.messageMetrics.Tx(header.RecvError, 0, 1)
|
f.messageMetrics.Tx(header.RecvError, 0, 1)
|
||||||
|
|
||||||
@@ -591,7 +594,7 @@ func (f *Interface) handleRecvError(addr netip.AddrPort, h *header.H) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
hr := hostinfo.GetRemote()
|
hr := hostinfo.GetRemote()
|
||||||
if hr.IsValid() && hr != addr {
|
if hr.IsValid() && hr != addr && !hostinfo.lanes.isLaneRemote(addr) {
|
||||||
f.l.Info("Someone spoofing recv_errors?",
|
f.l.Info("Someone spoofing recv_errors?",
|
||||||
"addr", addr,
|
"addr", addr,
|
||||||
"hostinfoRemote", hr,
|
"hostinfoRemote", hr,
|
||||||
|
|||||||
Reference in new issue
Block a user