From fb20de39b2db5e4c7f4940b3a4fabd8d562bce8f Mon Sep 17 00:00:00 2001 From: Wade Simmons Date: Wed, 2 Sep 2026 14:24:54 -0400 Subject: [PATCH] multiport: make a lane a session on the tunnel, not a tunnel Lanes used to be separate HostInfos, each with its own handshake, its own half-established states, its own lifetime and its own slot bookkeeping. That bought nothing: a lane is the same tunnel over a different underlay 5-tuple. Derive lane sessions instead. Noise leaves us with A.eKey == B.dKey, so both sides HKDF-expand the same two keys with the same per-lane label and land on a matched pair without exchanging anything. Lanes now cost no handshake, have no half-established state, and die exactly when their base tunnel does. Which lane a packet belongs to rides the low byte of the nebula header's Reserved field, inside the AEAD's associated data. Receiving on a lane needs no permission, since the session exists the moment the base handshake completes. Sending on one needs proof the new 5-tuple works, so a lane stays down until a probe on it is acked and falls back to the base tunnel the moment it stops being acked. Probing is demand-driven off the connection manager's per-tunnel traffic tick: a peer we exchange a trickle with never costs more than its base tunnel, however many lanes are configured. The ack rides the base session on purpose, so a broken reverse lane can't fail a working one. Because the data now rides lane counters, the rehandshake, exhaustion and swap-primary checks take the max counter across the base session and its lanes; otherwise the base counter would sit near zero while a lane ran its keys past the nonce ceiling. Removes OutboundLaneTimer, EnsureLanes, startLaneHandshake, completeLane, completeLaneResponder, makeLaneTrafficDecision and the lane fields on HostInfo. Handshake payload field 3 (the per-lane handshake index) is permanently reserved; peers advertise a TxLanes count instead. --- connection_manager.go | 96 +--- connection_state.go | 50 ++ control.go | 46 +- control_test.go | 2 +- handshake/machine.go | 13 +- handshake/machine_lanes_test.go | 34 +- handshake/payload.go | 23 +- handshake/payload_test.go | 8 +- handshake_manager.go | 457 ++----------------- header/header.go | 37 +- header/header_test.go | 26 +- hostmap.go | 277 +---------- inside.go | 75 ++- interface.go | 21 +- lanes.go | 462 +++++++++++++++++++ lanes_test.go | 781 ++++++++++++++++---------------- main.go | 7 + noiseutil/cipher_state.go | 27 ++ outside.go | 54 ++- 19 files changed, 1223 insertions(+), 1273 deletions(-) create mode 100644 lanes.go diff --git a/connection_manager.go b/connection_manager.go index 1712a6e2..06bc6402 100644 --- a/connection_manager.go +++ b/connection_manager.go @@ -197,17 +197,17 @@ func (cm *connectionManager) doTrafficCheck(localIndex uint32, p, nb, out []byte } cm.resetRelayTrafficCheck(hostinfo) - cm.ensureLanes(localIndex, decision, hostinfo) + cm.maintainLanes(localIndex, decision, hostinfo, now, nb, out) } -// ensureLanes piggybacks lane establishment on the per-tunnel traffic tick: -// any live base with lane state gets the slots its data plane asked for -// started (subject to the per-slot backoff). This tick is the right place for -// it precisely because lanes are demand-driven — a base only lands here when -// it has traffic, which is the same condition that raises lane demand. +// 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) ensureLanes(localIndex uint32, decision trafficDecision, hostinfo *HostInfo) { +func (cm *connectionManager) maintainLanes(localIndex uint32, decision trafficDecision, hostinfo *HostInfo, now time.Time, nb, out []byte) { if decision == deleteTunnel || decision == closeTunnel { return } @@ -217,10 +217,7 @@ func (cm *connectionManager) ensureLanes(localIndex uint32, decision trafficDeci return } } - if hostinfo.isLane() || hostinfo.lanes == nil { - return - } - cm.intf.handshakeManager.EnsureLanes(hostinfo) + cm.intf.probeLanes(hostinfo, now, nb, out) } func (cm *connectionManager) resetRelayTrafficCheck(hostinfo *HostInfo) { @@ -353,18 +350,15 @@ func (cm *connectionManager) makeTrafficDecision(localIndex uint32, now time.Tim return closeTunnel, hostinfo, nil } - // Checked ahead of the lane branch: an exhausted counter is fatal for lanes too, and makeLaneTrafficDecision - // only reasons about liveness. - if hostinfo.ConnectionState != nil && hostinfo.ConnectionState.messageCounter.Load() >= RejectAfterMessages { + // The highest counter across the base session and its lanes: the lanes carry + // the data, so the base counter alone would sit near zero while a lane runs + // its keys past the nonce ceiling. + if hostinfo.maxMessageCounter() >= RejectAfterMessages { // Send path can't encrypt a CloseTunnel notify, so just delete locally; the peer recovers via recv_error. hostinfo.logger(cm.l).Error("Dropping tunnel, message counter is exhausted") return deleteTunnel, hostinfo, nil } - if hostinfo.isLane() { - return cm.makeLaneTrafficDecision(hostinfo, now) - } - primary := cm.hostMap.Hosts[hostinfo.vpnAddrs[0]] mainHostInfo := true if primary != nil && primary != hostinfo { @@ -461,72 +455,14 @@ func (cm *connectionManager) makeTrafficDecision(localIndex uint32, now time.Tim return decision, hostinfo, nil } -// makeLaneTrafficDecision is the lane-tunnel subset of makeTrafficDecision: -// no primary/swap/rehandshake logic (lanes are never primary), no punches -// (lane keepalives egress the lane's own socket and are its NAT keepalive), -// just alive / test / dead. A dead lane's DeleteHostInfo clears its base slot -// with backoff, and ensureLanes re-establishes it. -func (cm *connectionManager) makeLaneTrafficDecision(hostinfo *HostInfo, now time.Time) (trafficDecision, *HostInfo, *HostInfo) { - inTraffic, _ := cm.getAndResetTrafficCheck(hostinfo, now) - - if inTraffic { - if cm.l.Enabled(context.Background(), slog.LevelDebug) { - hostinfo.logger(cm.l).Debug("Tunnel status", - "tunnelCheck", m{"state": "alive", "method": "passive"}, - "laneIndex", hostinfo.laneIndex, - ) - } - hostinfo.setPendingDeletion(false) - cm.trafficTimer.Add(hostinfo.localIndexId, cm.checkInterval) - return doNothing, hostinfo, nil - } - - if hostinfo.isPendingDeletion() { - hostinfo.logger(cm.l).Info("Tunnel status", - "tunnelCheck", m{"state": "dead", "method": "active"}, - "laneIndex", hostinfo.laneIndex, - ) - return deleteTunnel, hostinfo, nil - } - - // Idle lanes are actively kept alive (unlike idle base tunnels, which - // just get punches): a lane is datapath infrastructure and its keepalive - // doubles as the per-path death detector. - decision := doNothing - if hostinfo.ConnectionState != nil { - decision = sendTestPacket - } - hostinfo.setPendingDeletion(true) - cm.trafficTimer.Add(hostinfo.localIndexId, cm.pendingDeletionInterval) - return decision, hostinfo, nil -} - func (cm *connectionManager) isInactive(hostinfo *HostInfo, now time.Time) (time.Duration, bool) { if cm.dropInactive.Load() == false { // We aren't configured to drop inactive tunnels return 0, false } - // With multiport the data rides the lanes and the base may look idle; - // a base is only inactive if its whole lane family is. lastUsed is only - // written by this ticker goroutine, so these reads are safe. - lastUsed := hostinfo.lastUsed - if ls := hostinfo.lanes; ls != nil { - for i := range ls.txLanes { - if lane := ls.txLanes[i].Load(); lane != nil && lane.lastUsed.After(lastUsed) { - lastUsed = lane.lastUsed - } - } - ls.Lock() - for _, lane := range ls.peerLanes { - if lane.lastUsed.After(lastUsed) { - lastUsed = lane.lastUsed - } - } - ls.Unlock() - } - - inactiveDuration := now.Sub(lastUsed) + // Lane traffic is this hostinfo's traffic, so lastUsed already covers it. + inactiveDuration := now.Sub(hostinfo.lastUsed) if inactiveDuration < cm.getInactivityTimeout() { // It's not considered inactive return inactiveDuration, false @@ -549,7 +485,7 @@ func (cm *connectionManager) shouldSwapPrimary(current *HostInfo) bool { return false } - if current.ConnectionState.messageCounter.Load() >= RehandshakeAfterMessages { + if current.maxMessageCounter() >= RehandshakeAfterMessages { // This tunnel is being rolled for counter exhaustion, never swap back onto its spent key. return false } @@ -653,7 +589,7 @@ func (cm *connectionManager) tryRehandshake(hostinfo *HostInfo) { cm.intf.handshakeManager.StartHandshake(hostinfo.vpnAddrs[0], nil) return } - if hostinfo.ConnectionState.messageCounter.Load() >= RehandshakeAfterMessages { + if hostinfo.maxMessageCounter() >= RehandshakeAfterMessages { cm.l.Info("Re-handshaking with remote", "vpnAddrs", hostinfo.vpnAddrs, "reason", "message counter rehandshake threshold reached", diff --git a/connection_state.go b/connection_state.go index 3018048b..8f5ef52a 100644 --- a/connection_state.go +++ b/connection_state.go @@ -1,9 +1,12 @@ package nebula import ( + "crypto/hkdf" + "crypto/sha256" "encoding/json" "fmt" "log/slog" + "strconv" "sync" "sync/atomic" @@ -74,6 +77,53 @@ func newConnectionStateFromResult(r *handshake.Result) (*ConnectionState, error) return ci, nil } +// newLaneConnectionState derives multiport lane s's session from the completed +// base handshake. Each key is an HKDF expansion of the base tunnel's matching +// key, labelled with the lane index, so the pair stays matched with no extra +// negotiation: Noise leaves our send key equal to the peer's receive key, and +// expanding both with the same label preserves that. +// +// The lane gets its own counter and replay window starting from zero. No +// handshake messages were spent on it, so unlike the base session there is +// nothing to seed. +func newLaneConnectionState(r *handshake.Result, lane uint8) (*ConnectionState, error) { + if lane == 0 { + return nil, fmt.Errorf("lane 0 is the base session") + } + + eKey, err := deriveLaneKey(r.EKey.UnsafeKey(), lane) + if err != nil { + return nil, err + } + dKey, err := deriveLaneKey(r.DKey.UnsafeKey(), lane) + if err != nil { + return nil, err + } + + return &ConnectionState{ + myCert: r.MyCert, + initiator: r.Initiator, + peerCert: r.RemoteCert, + eKey: noiseutil.NewCipherStateFromKey(eKey, r.Cipher), + dKey: noiseutil.NewCipherStateFromKey(dKey, r.Cipher), + window: NewBits(ReplayWindow), + epoch: sessionEpoch.Add(1), + }, nil +} + +// deriveLaneKey expands a base tunnel key into the key for one lane. +func deriveLaneKey(base [32]byte, lane uint8) ([32]byte, error) { + var out [32]byte + // The base key is already unique to this tunnel and direction, so the lane + // index is the only thing that needs to vary; no salt is required. + k, err := hkdf.Key(sha256.New, base[:], nil, laneKeyInfo+" "+strconv.Itoa(int(lane)), len(out)) + if err != nil { + return out, err + } + copy(out[:], k) + return out, nil +} + func (cs *ConnectionState) MarshalJSON() ([]byte, error) { return json.Marshal(m{ "certificate": cs.peerCert, diff --git a/control.go b/control.go index e0554e68..9928657a 100644 --- a/control.go +++ b/control.go @@ -67,9 +67,16 @@ type ControlHostInfo struct { CurrentRemote netip.AddrPort `json:"currentRemote"` CurrentRelaysToMe []netip.Addr `json:"currentRelaysToMe"` CurrentRelaysThroughMe []netip.Addr `json:"currentRelaysThroughMe"` - IsLane bool `json:"isLane,omitempty"` - LaneIndex uint16 `json:"laneIndex,omitempty"` - SockIdx int `json:"sockIdx,omitempty"` + Lanes []ControlLane `json:"lanes,omitempty"` +} + +// ControlLane reports one multiport lane of a tunnel. Only lanes we may send on +// are listed; receive-only lanes have no state worth showing. +type ControlLane struct { + Index uint8 `json:"index"` + Up bool `json:"up"` + Remote netip.AddrPort `json:"remote,omitempty"` + MessageCounter uint64 `json:"messageCounter"` } // Start actually runs nebula, this is a nonblocking call. @@ -365,11 +372,6 @@ func (c *Control) CloseAllTunnels(excludeLighthouses bool) (closed int) { // Grab the hostMap lock to access the Hosts map c.f.hostMap.Lock() for _, relayHost := range c.f.hostMap.Indexes { - // Lanes ride along with their base tunnel's shutdown cascade; closing - // them individually would race the cascade's identity-checked deletes. - if relayHost.isLane() { - continue - } if _, ok := relayingHosts[relayHost.vpnAddrs[0]]; !ok { hostInfos = append(hostInfos, relayHost) } @@ -398,9 +400,7 @@ func copyHostInfo(h *HostInfo, preferredRanges []netip.Prefix) ControlHostInfo { CurrentRelaysToMe: h.relayState.CopyRelayIps(), CurrentRelaysThroughMe: h.relayState.CopyRelayForIps(), CurrentRemote: h.GetRemote(), - IsLane: h.isLane(), - LaneIndex: h.laneIndex, - SockIdx: h.sockIdx, + Lanes: copyLanes(h), } for i, a := range h.vpnAddrs { @@ -418,6 +418,30 @@ func copyHostInfo(h *HostInfo, preferredRanges []netip.Prefix) ControlHostInfo { return chi } +// copyLanes snapshots the sendable multiport lanes of a tunnel, or nil when it +// has none. txAddr is the lane's gate as well as its destination, so a nil load +// is exactly "this lane is down and its routine is riding the base tunnel". +func copyLanes(h *HostInfo) []ControlLane { + ls := h.lanes + if ls == nil || ls.txLanes < 2 { + return nil + } + + lanes := make([]ControlLane, 0, ls.txLanes-1) + for s := 1; s < ls.txLanes; s++ { + l := ControlLane{Index: uint8(s)} + if addr := ls.txAddr[s].Load(); addr != nil { + l.Up = true + l.Remote = *addr + } + if cs := ls.sessions[s]; cs != nil { + l.MessageCounter = cs.messageCounter.Load() + } + lanes = append(lanes, l) + } + return lanes +} + func listHostMapHosts(hl controlHostLister) []ControlHostInfo { hosts := make([]ControlHostInfo, 0) pr := hl.GetPreferredRanges() diff --git a/control_test.go b/control_test.go index fb0a79ed..c32e7be0 100644 --- a/control_test.go +++ b/control_test.go @@ -105,7 +105,7 @@ func TestControl_GetHostInfoByVpnIp(t *testing.T) { } // Make sure we don't have any unexpected fields - assertFields(t, []string{"VpnAddrs", "LocalIndex", "RemoteIndex", "RemoteAddrs", "Cert", "MessageCounter", "CurrentRemote", "CurrentRelaysToMe", "CurrentRelaysThroughMe", "IsLane", "LaneIndex", "SockIdx"}, thi) + assertFields(t, []string{"VpnAddrs", "LocalIndex", "RemoteIndex", "RemoteAddrs", "Cert", "MessageCounter", "CurrentRemote", "CurrentRelaysToMe", "CurrentRelaysThroughMe", "Lanes"}, thi) assert.Equal(t, &expectedInfo, thi) test.AssertDeepCopyEqual(t, &expectedInfo, thi) diff --git a/handshake/machine.go b/handshake/machine.go index 50b29bdc..ab82fee9 100644 --- a/handshake/machine.go +++ b/handshake/machine.go @@ -42,10 +42,11 @@ type Result struct { // Multiport lane negotiation, from the peer's LaneDetails. All zero when // the peer did not advertise (vanilla peer or multiport disabled). - // PeerLaneIndex is nonzero only on the responder side of a lane handshake. + // PeerTxLanes is how many lanes the peer may send on, which is how many lane + // sessions we need in order to receive everything it sends. PeerPortCount uint32 PeerBasePort uint32 - PeerLaneIndex uint32 + PeerTxLanes uint32 } // Machine drives a Noise handshake through N messages. It handles Noise @@ -344,18 +345,18 @@ func (m *Machine) processPayload(msg []byte, flags msgFlags) error { // Multiport advert from the peer's side of the exchange. Out-of-range // values mean a peer we can't pair lanes with; ignore the advert // rather than failing the handshake — the tunnel itself is fine, it - // just won't get lanes. Semantic policing (index bounds vs advert, - // port-count caps) belongs to the handshake manager. + // just won't get lanes. Semantic policing (port-count caps, lane + // clamping) belongs to the handshake manager. var peerLanes *LaneDetails if m.result.Initiator { peerLanes = payload.ResponderLanes } else { peerLanes = payload.InitiatorLanes } - if peerLanes != nil && peerLanes.BasePort <= 0xffff && peerLanes.PortCount <= 0xffff { + if peerLanes != nil && peerLanes.BasePort <= 0xffff && peerLanes.PortCount <= 0xffff && peerLanes.TxLanes <= 0xffff { m.result.PeerPortCount = peerLanes.PortCount m.result.PeerBasePort = peerLanes.BasePort - m.result.PeerLaneIndex = peerLanes.LaneIndex + m.result.PeerTxLanes = peerLanes.TxLanes } } diff --git a/handshake/machine_lanes_test.go b/handshake/machine_lanes_test.go index ef6c0c6b..3446afdf 100644 --- a/handshake/machine_lanes_test.go +++ b/handshake/machine_lanes_test.go @@ -61,30 +61,18 @@ func doFullLaneHandshake(t *testing.T, initLanes, respLanes *LaneDetails) (initR func TestMachineLaneAdvertBothSides(t *testing.T) { initR, respR := doFullLaneHandshake(t, - &LaneDetails{PortCount: 8, BasePort: 4242}, - &LaneDetails{PortCount: 4, BasePort: 5353}, + &LaneDetails{PortCount: 8, BasePort: 4242, TxLanes: 8}, + &LaneDetails{PortCount: 4, BasePort: 5353, TxLanes: 3}, ) // Each side's Result carries the peer's advert. assert.Equal(t, uint32(4), initR.PeerPortCount) assert.Equal(t, uint32(5353), initR.PeerBasePort) - assert.Equal(t, uint32(0), initR.PeerLaneIndex) + assert.Equal(t, uint32(3), initR.PeerTxLanes) assert.Equal(t, uint32(8), respR.PeerPortCount) assert.Equal(t, uint32(4242), respR.PeerBasePort) - assert.Equal(t, uint32(0), respR.PeerLaneIndex) -} - -func TestMachineLaneHandshakeCarriesLaneIndex(t *testing.T) { - // A lane handshake: initiator tags its lane number; responder still - // adverts (harmlessly). - initR, respR := doFullLaneHandshake(t, - &LaneDetails{PortCount: 8, BasePort: 4242, LaneIndex: 3}, - &LaneDetails{PortCount: 4, BasePort: 5353}, - ) - - assert.Equal(t, uint32(3), respR.PeerLaneIndex) - assert.Equal(t, uint32(0), initR.PeerLaneIndex) + assert.Equal(t, uint32(8), respR.PeerTxLanes) } func TestMachineLaneAdvertAsymmetric(t *testing.T) { @@ -103,11 +91,21 @@ func TestMachineLaneAdvertAsymmetric(t *testing.T) { func TestMachineLaneAdvertOutOfRangeIgnored(t *testing.T) { // A BasePort that can't be a real UDP port is ignored, not fatal. initR, respR := doFullLaneHandshake(t, - &LaneDetails{PortCount: 8, BasePort: 70000}, - &LaneDetails{PortCount: 4, BasePort: 5353}, + &LaneDetails{PortCount: 8, BasePort: 70000, TxLanes: 8}, + &LaneDetails{PortCount: 4, BasePort: 5353, TxLanes: 4}, ) assert.Equal(t, uint32(0), respR.PeerPortCount) assert.Equal(t, uint32(0), respR.PeerBasePort) + assert.Equal(t, uint32(0), respR.PeerTxLanes) // The sane side still negotiates. assert.Equal(t, uint32(4), initR.PeerPortCount) + + // An out-of-range TxLanes drops the whole advert the same way: a lane index + // that does not fit the header is as unusable as an impossible port. + initR, respR = doFullLaneHandshake(t, + &LaneDetails{PortCount: 8, BasePort: 4242, TxLanes: 0x10000}, + &LaneDetails{PortCount: 4, BasePort: 5353, TxLanes: 4}, + ) + assert.Equal(t, uint32(0), respR.PeerPortCount) + assert.Equal(t, uint32(4), initR.PeerPortCount) } diff --git a/handshake/payload.go b/handshake/payload.go index 8adec998..9d8e1574 100644 --- a/handshake/payload.go +++ b/handshake/payload.go @@ -27,12 +27,14 @@ type Payload struct { ResponderLanes *LaneDetails } -// LaneDetails advertises multiport lane capability. LaneIndex is zero on base -// handshakes and the initiator's lane number (>= 1) on lane handshakes. +// LaneDetails advertises multiport lane capability: the contiguous UDP port +// range the sender bound, and how many lanes it may send on. The receiver needs +// PortCount/BasePort to aim its own lanes and TxLanes to know how many lane +// sessions to derive for receiving. type LaneDetails struct { PortCount uint32 BasePort uint32 - LaneIndex uint32 + TxLanes uint32 } // Proto field numbers for NebulaHandshakeDetails @@ -46,11 +48,12 @@ const ( fieldResponderLanes = 10 // LaneDetails ) -// Proto field numbers for LaneDetails +// Proto field numbers for LaneDetails. +// Field 3 was a per-lane handshake index and is permanently reserved. const ( fieldLanePortCount = 1 // uint32 fieldLaneBasePort = 2 // uint32 - fieldLaneLaneIndex = 3 // uint32 + fieldLaneTxLanes = 4 // uint32 ) // MarshalPayload encodes a handshake payload in protobuf wire format compatible @@ -105,8 +108,8 @@ func (d *LaneDetails) marshal(out []byte) []byte { out = protowire.AppendVarint(out, uint64(d.PortCount)) out = protowire.AppendTag(out, fieldLaneBasePort, protowire.VarintType) out = protowire.AppendVarint(out, uint64(d.BasePort)) - out = protowire.AppendTag(out, fieldLaneLaneIndex, protowire.VarintType) - out = protowire.AppendVarint(out, uint64(d.LaneIndex)) + out = protowire.AppendTag(out, fieldLaneTxLanes, protowire.VarintType) + out = protowire.AppendVarint(out, uint64(d.TxLanes)) return out } @@ -256,7 +259,7 @@ func unmarshalLaneDetails(d *LaneDetails, b []byte) error { // wire-type mismatch, unknown fields are skipped, repeated singular // fields follow proto3 last-wins. switch num { - case fieldLanePortCount, fieldLaneBasePort, fieldLaneLaneIndex: + case fieldLanePortCount, fieldLaneBasePort, fieldLaneTxLanes: if typ != protowire.VarintType { return errInvalidHandshakeDetails } @@ -269,8 +272,8 @@ func unmarshalLaneDetails(d *LaneDetails, b []byte) error { d.PortCount = uint32(v) case fieldLaneBasePort: d.BasePort = uint32(v) - case fieldLaneLaneIndex: - d.LaneIndex = uint32(v) + case fieldLaneTxLanes: + d.TxLanes = uint32(v) } b = b[n:] default: diff --git a/handshake/payload_test.go b/handshake/payload_test.go index 73899c64..bd9bc677 100644 --- a/handshake/payload_test.go +++ b/handshake/payload_test.go @@ -146,14 +146,14 @@ func TestPayloadLaneDetails(t *testing.T) { data := MarshalPayload(nil, Payload{ InitiatorIndex: 12345, Time: 999, - InitiatorLanes: &LaneDetails{PortCount: 8, BasePort: 4242, LaneIndex: 3}, + InitiatorLanes: &LaneDetails{PortCount: 8, BasePort: 4242, TxLanes: 3}, ResponderLanes: &LaneDetails{PortCount: 4, BasePort: 5353}, }) got, err := UnmarshalPayload(data) require.NoError(t, err) require.NotNil(t, got.InitiatorLanes) - assert.Equal(t, LaneDetails{PortCount: 8, BasePort: 4242, LaneIndex: 3}, *got.InitiatorLanes) + assert.Equal(t, LaneDetails{PortCount: 8, BasePort: 4242, TxLanes: 3}, *got.InitiatorLanes) require.NotNil(t, got.ResponderLanes) assert.Equal(t, LaneDetails{PortCount: 4, BasePort: 5353}, *got.ResponderLanes) }) @@ -237,7 +237,7 @@ func TestPayloadLaneDetails(t *testing.T) { t.Run("lane subfield varint overflow rejected", func(t *testing.T) { var lane []byte - lane = protowire.AppendTag(lane, fieldLaneLaneIndex, protowire.VarintType) + lane = protowire.AppendTag(lane, fieldLaneTxLanes, protowire.VarintType) lane = protowire.AppendVarint(lane, math.MaxUint32+1) var details []byte @@ -442,7 +442,7 @@ func FuzzPayload(f *testing.F) { f.Add(MarshalPayload(nil, Payload{ InitiatorIndex: 1, Time: 3, - InitiatorLanes: &LaneDetails{PortCount: 8, BasePort: 4242, LaneIndex: 2}, + InitiatorLanes: &LaneDetails{PortCount: 8, BasePort: 4242, TxLanes: 2}, ResponderLanes: &LaneDetails{PortCount: 4, BasePort: 5353}, })) f.Add([]byte{}) diff --git a/handshake_manager.go b/handshake_manager.go index 93ab61db..4f65eeb6 100644 --- a/handshake_manager.go +++ b/handshake_manager.go @@ -52,7 +52,7 @@ type HandshakeConfig struct { // Multiport lane parameters; laneCount == 0 means multiport is disabled. // laneCount includes implicit lane 0 (the base tunnel), so lanes - // 1..laneCount-1 are initiated. lanePortCount/laneBasePort describe our + // 1..laneCount-1 may carry traffic. lanePortCount/laneBasePort describe our // own bound port range and are advertised in every handshake payload. laneCount int lanePortCount uint16 @@ -73,15 +73,11 @@ type HandshakeManager struct { outside udp.Conn config HandshakeConfig OutboundHandshakeTimer *LockingTimerWheel[netip.Addr] - // OutboundLaneTimer drives lane handshake retries. Lanes never enter - // vpnIps (they would collide with base handshakes for the same address), - // so their wheel is keyed by pending localIndexId instead. - OutboundLaneTimer *LockingTimerWheel[uint32] - messageMetrics *MessageMetrics - metricInitiated metrics.Counter - metricTimedOut metrics.Counter - f *Interface - l *slog.Logger + messageMetrics *MessageMetrics + metricInitiated metrics.Counter + metricTimedOut metrics.Counter + f *Interface + l *slog.Logger // can be used to trigger outbound handshake for the given vpnIp trigger chan netip.Addr @@ -100,11 +96,6 @@ type HandshakeHostInfo struct { hostinfo *HostInfo machine *handshake.Machine // The handshake state machine, set during stage 0 (initiator) or beginHandshake (responder multi-message) - - // laneTarget is the single pinned destination for a lane handshake - // (base-tunnel remote IP at the peer's lane port). Lane handshakes never - // broadcast to the RemoteList. - laneTarget netip.AddrPort } func (hh *HandshakeHostInfo) cachePacket(l *slog.Logger, t header.MessageType, st header.MessageSubType, packet []byte, f packetCallback, m *cachedPacketMetrics) { @@ -142,7 +133,6 @@ func NewHandshakeManager(l *slog.Logger, mainHostMap *HostMap, lightHouse *Light config: config, trigger: make(chan netip.Addr, config.triggerBuffer), OutboundHandshakeTimer: NewLockingTimerWheel[netip.Addr](config.tryInterval, hsTimeout(config.retries, config.tryInterval)), - OutboundLaneTimer: NewLockingTimerWheel[uint32](config.tryInterval, hsTimeout(config.retries, config.tryInterval)), messageMetrics: config.messageMetrics, metricInitiated: metrics.GetOrRegisterCounter("handshake_manager.initiated", nil), metricTimedOut: metrics.GetOrRegisterCounter("handshake_manager.timed_out", nil), @@ -162,7 +152,6 @@ func (hm *HandshakeManager) Run(ctx context.Context) { hm.handleOutbound(vpnIP, true) case now := <-clockSource.C: hm.NextOutboundHandshakeTimerTick(now) - hm.NextOutboundLaneTimerTick(now) } } } @@ -554,10 +543,9 @@ func (hm *HandshakeManager) DeleteHostInfo(hostinfo *HostInfo) { func (hm *HandshakeManager) unlockedDeleteHostInfo(hostinfo *HostInfo) { for _, addr := range hostinfo.vpnAddrs { - // Only delete the pending entry if it is actually ours. Lane - // handshakes never live in vpnIps, and an unconditional delete here - // could evict a concurrently pending base handshake for the same - // address. + // Only delete the pending entry if it is actually ours: an + // unconditional delete could evict a concurrently pending handshake for + // the same address. if cur, ok := hm.vpnIps[addr]; ok && cur.hostinfo == hostinfo { delete(hm.vpnIps, addr) } @@ -695,7 +683,7 @@ func (hm *HandshakeManager) buildStage0Packet(hh *HandshakeHostInfo) bool { v, cs.GetCredential, hm.certVerifier(), func() (uint32, error) { return hm.allocateIndex(hh) }, true, header.HandshakeIXPSK0, - hm.laneAdvert(uint32(hh.hostinfo.laneIndex)), + hm.laneAdvert(), ) if err != nil { hm.f.l.Error("Failed to create handshake machine", @@ -721,227 +709,38 @@ func (hm *HandshakeManager) buildStage0Packet(hh *HandshakeHostInfo) bool { // laneAdvert returns our multiport advert for a handshake payload, or nil // when multiport is disabled (which keeps the payload byte-identical to -// vanilla). laneIndex is 0 for base handshakes and our lane number for lane -// handshakes. -func (hm *HandshakeManager) laneAdvert(laneIndex uint32) *handshake.LaneDetails { +// vanilla). +func (hm *HandshakeManager) laneAdvert() *handshake.LaneDetails { if hm.config.laneCount == 0 { return nil } return &handshake.LaneDetails{ PortCount: uint32(hm.config.lanePortCount), BasePort: uint32(hm.config.laneBasePort), - LaneIndex: laneIndex, + TxLanes: uint32(hm.config.laneCount), } } -// maybeAllocLaneState attaches a laneState to a just-completed base tunnel -// when both sides advertised multiport. Must run before the hostinfo becomes -// visible in the hostmap: the data plane reads base.lanes lock-free. -func (hm *HandshakeManager) maybeAllocLaneState(hostinfo *HostInfo, result *handshake.Result) { - if hm.config.laneCount == 0 || result.PeerPortCount == 0 || result.PeerLaneIndex != 0 { - return - } - peerPorts := result.PeerPortCount - if peerPorts > 256 { - // A certified-but-hostile peer doesn't get to size our state. - peerPorts = 256 - } - var offset uint16 - if len(hm.f.myVpnAddrs) > 0 && len(hostinfo.vpnAddrs) > 0 { - offset = lanePortOffset(hm.f.myVpnAddrs[0], hostinfo.vpnAddrs[0], uint16(peerPorts)) - } - hostinfo.lanes = newLaneState(hm.config.laneCount, uint16(peerPorts), uint16(result.PeerBasePort), offset) -} - -// EnsureLanes starts lane handshakes for every empty, non-pending, retry-due -// slot of a base tunnel that the data plane has asked for. Called on base -// handshake completion (both sides) and from the connection manager's -// per-tunnel tick, so a demanded lane is established, and a dead one -// re-established, within one check interval subject to per-slot backoff. +// maybeAllocLanes derives the multiport lane sessions for a just-completed +// tunnel. Must run before the hostinfo becomes visible in the hostmap: the data +// plane reads hostinfo.lanes with no synchronization at all. // -// Demand is what makes lanes lazy: nothing is built when the base tunnel -// completes, only when a routine actually has traffic for the peer and finds -// its slot empty (see noteLaneDemand). Consuming the flag here rather than -// leaving it set also means a lane that dies after its routine went quiet -// stays dead instead of being rebuilt forever. -func (hm *HandshakeManager) EnsureLanes(base *HostInfo) { - ls := base.lanes - if ls == nil || hm.config.laneCount <= 1 { +// A failure here costs the tunnel its lanes, not the tunnel: the base session is +// already usable and every lane the data plane wants falls back to it. +func (hm *HandshakeManager) maybeAllocLanes(hostinfo *HostInfo, result *handshake.Result) { + if hm.config.laneCount == 0 || result.PeerPortCount == 0 { + return + } + if len(hm.f.myVpnAddrs) == 0 || len(hostinfo.vpnAddrs) == 0 { return } - now := time.Now() - var starts []int - ls.Lock() - n := min(len(ls.txLanes), hm.config.laneCount) - for i := 1; i < n; i++ { - if ls.txLanes[i].Load() != nil || ls.txPending[i] || now.Before(ls.txRetryAt[i]) { - continue - } - // Checked last: a slot that isn't startable keeps its demand for the - // tick that can act on it. - if !ls.takeLaneDemand(i) { - continue - } - ls.txPending[i] = true - starts = append(starts, i) - } - ls.Unlock() - - for _, i := range starts { - hm.startLaneHandshake(base, i) - } -} - -// startLaneHandshake initiates lane i of base: a full Noise handshake from -// local socket i to the peer's advertised lane port. The caller has already -// claimed the slot (txPending[i] = true); every failure path must release it -// via noteLaneFailure. -func (hm *HandshakeManager) startLaneHandshake(base *HostInfo, i int) { - ls := base.lanes - - remote := base.GetRemote() - if !remote.IsValid() || ls.peerPortCount == 0 { - // Relay-only peer (or a zero advert that should not have allocated - // lanes). Retried if a direct path shows up later. - ls.noteLaneFailure(i) - return - } - target := netip.AddrPortFrom(remote.Addr(), ls.laneTargetPort(i)) - - hostinfo := &HostInfo{ - vpnAddrs: slices.Clone(base.vpnAddrs), - HandshakePacket: make(map[uint8][]byte, 0), - relayState: RelayState{ - relays: nil, - relayForByAddr: map[netip.Addr]*Relay{}, - relayForByIdx: map[uint32]*Relay{}, - }, - // A private RemoteList: SetRemote must never leak lane ports into the - // shared lighthouse-learned cache that base handshakes broadcast to. - remotes: NewRemoteList(base.vpnAddrs, nil), - sockIdx: i, - laneIndex: uint16(i), - laneOwned: true, - parent: base, - } - - hh := &HandshakeHostInfo{ - hostinfo: hostinfo, - startTime: time.Now(), - laneTarget: target, - } - if cs := base.ConnectionState; cs != nil && cs.myCert != nil { - // Pin the lane to the base tunnel's negotiated cert version rather - // than re-running version selection. - hh.initiatingVersionOverride = cs.myCert.Version() - } - - // Build stage 0 eagerly: allocateIndex registers hh in hm.indexes so - // continueHandshake can find it, and gives us the key for the lane timer. - hh.Lock() - ok := hm.buildStage0Packet(hh) - hh.Unlock() - if !ok { - ls.noteLaneFailure(i) - return - } - - hostinfo.logger(hm.l).Info("Lane handshake started", - "laneIndex", i, - "udpAddr", target, - "initiatorIndex", hostinfo.localIndexId, - ) - hm.metricInitiated.Inc(1) - hm.OutboundLaneTimer.Add(hostinfo.localIndexId, hm.config.tryInterval) - hm.handleOutboundLane(hostinfo.localIndexId) -} - -func (hm *HandshakeManager) NextOutboundLaneTimerTick(now time.Time) { - hm.OutboundLaneTimer.Advance(now) - for { - idx, has := hm.OutboundLaneTimer.Purge() - if !has { - break - } - hm.handleOutboundLane(idx) - } -} - -// handleOutboundLane is handleOutbound for lane handshakes: single pinned -// target, egress via the lane's own socket, no lighthouse and no relays. -func (hm *HandshakeManager) handleOutboundLane(localIndex uint32) { - hh := hm.queryIndex(localIndex) - if hh == nil || !hh.hostinfo.isLane() { - return - } - hh.Lock() - defer hh.Unlock() - - hostinfo := hh.hostinfo - if hh.counter >= hm.config.retries { - hostinfo.logger(hm.l).Info("Lane handshake timed out", - "laneIndex", hostinfo.laneIndex, - "udpAddr", hh.laneTarget, - "initiatorIndex", hostinfo.localIndexId, - "durationNs", time.Since(hh.startTime).Nanoseconds(), - ) - hm.metricTimedOut.Inc(1) - hm.DeleteHostInfo(hostinfo) - hostinfo.parent.lanes.noteLaneFailure(int(hostinfo.laneIndex)) - return - } - hh.counter++ - - stage0 := hostinfo.HandshakePacket[handshakePacketStage0] - hm.messageMetrics.Tx(header.Handshake, hh.machine.Subtype(), 1) - err := hm.f.writers[hostinfo.sockIdx].WriteTo(stage0, hh.laneTarget) + ls, err := newLaneSet(result, hm.config.laneCount, hm.f.myVpnAddrs[0], hostinfo.vpnAddrs[0]) if err != nil { - hostinfo.logger(hm.l).Error("Failed to send lane handshake message", - "laneIndex", hostinfo.laneIndex, - "udpAddr", hh.laneTarget, - "initiatorIndex", hostinfo.localIndexId, - "error", err, - ) - } else if hm.l.Enabled(context.Background(), slog.LevelDebug) { - hostinfo.logger(hm.l).Debug("Lane handshake message sent", - "laneIndex", hostinfo.laneIndex, - "udpAddr", hh.laneTarget, - "initiatorIndex", hostinfo.localIndexId, - ) + hostinfo.logger(hm.l).Error("Failed to derive multiport lanes", "error", err) + return } - - hm.OutboundLaneTimer.Add(localIndex, hm.config.tryInterval*time.Duration(hh.counter)) -} - -// deletePendingHostInfo abandons a pending handshake. For lanes it also -// releases the base's slot claim with failure backoff so ensureLanes can -// retry later. -func (hm *HandshakeManager) deletePendingHostInfo(hostinfo *HostInfo) { - hm.DeleteHostInfo(hostinfo) - if hostinfo.isLane() { - hostinfo.parent.lanes.noteLaneFailure(int(hostinfo.laneIndex)) - } -} - -// completeLane moves a finished lane out of the pending map and registers it -// in the main hostmap's index maps (never Hosts). The caller publishes it to -// the base's txLanes/peerLanes afterwards. -func (hm *HandshakeManager) completeLane(hostinfo *HostInfo, f *Interface) { - hm.mainHostMap.Lock() - defer hm.mainHostMap.Unlock() - hm.Lock() - defer hm.Unlock() - - existingRemoteIndex, found := hm.mainHostMap.RemoteIndexes[hostinfo.remoteIndexId] - if found && existingRemoteIndex != nil { - hostinfo.logger(hm.l).Info("New lane shadows existing host remoteIndex", - "collision", existingRemoteIndex.vpnAddrs, - ) - } - - hm.unlockedDeleteHostInfo(hostinfo) - hm.mainHostMap.unlockedAddLane(hostinfo, f) + hostinfo.lanes = ls } // beginHandshake handles an incoming handshake packet that doesn't match any @@ -962,7 +761,7 @@ func (hm *HandshakeManager) beginHandshake(via ViaSender, packet []byte, h *head v, cs.GetCredential, hm.certVerifier(), func() (uint32, error) { return generateIndex(f.l) }, false, header.HandshakeIXPSK0, - hm.laneAdvert(0), + hm.laneAdvert(), ) if err != nil { f.l.Error("Failed to create handshake machine", "from", via, "error", err) @@ -999,13 +798,6 @@ func (hm *HandshakeManager) beginHandshake(via ViaSender, packet []byte, h *head return } - // A nonzero lane tag makes this a lane handshake for an existing base - // tunnel rather than a (possibly duplicate) tunnel of its own. - if result.PeerLaneIndex > 0 { - hm.completeLaneResponder(via, packet, response, result, vpnAddrs) - return - } - connState, err := newConnectionStateFromResult(result) if err != nil { f.l.Error("Discarding handshake with an invalid message index", "error", err, "vpnAddrs", vpnAddrs) @@ -1056,7 +848,7 @@ func (hm *HandshakeManager) beginHandshake(via ViaSender, packet []byte, h *head hostinfo.SetRemote(via.UdpAddr) } hostinfo.buildNetworks(f.myVpnNetworksTable, remoteCert.Certificate) - hm.maybeAllocLaneState(hostinfo, result) + hm.maybeAllocLanes(hostinfo, result) existing, err := hm.CheckAndComplete(hostinfo, handshakePacketStage0, f) if err != nil { @@ -1066,7 +858,6 @@ func (hm *HandshakeManager) beginHandshake(via ViaSender, packet []byte, h *head hm.sendHandshakeResponse(via, response, hostinfo, false) hostinfo.remotes.RefreshFromHandshake(vpnAddrs) - hm.EnsureLanes(hostinfo) // Don't wait for UpdateWorker if f.lightHouse.IsAnyLighthouseAddr(vpnAddrs) { @@ -1074,146 +865,6 @@ func (hm *HandshakeManager) beginHandshake(via ViaSender, packet []byte, h *head } } -// completeLaneResponder finishes the responder side of a lane handshake: -// associate with the base tunnel by the peer's certified vpn address, dedup -// replays per lane, register the lane in the index maps, and reply from the -// arrival socket. Handshakes with no live base are dropped — the initiator's -// retry loop converges once the base exists. -func (hm *HandshakeManager) completeLaneResponder(via ViaSender, packet, response []byte, result *handshake.Result, vpnAddrs []netip.Addr) { - f := hm.f - - if via.IsRelayed { - hm.l.Debug("dropping relayed lane handshake", "vpnAddrs", vpnAddrs, "from", via) - return - } - if hm.config.laneCount == 0 { - hm.l.Debug("dropping lane handshake, multiport is disabled", "vpnAddrs", vpnAddrs, "from", via) - return - } - - base := hm.mainHostMap.QueryVpnAddr(vpnAddrs[0]) - if base == nil || base.ConnectionState == nil || base.lanes == nil { - hm.l.Debug("dropping lane handshake with no base tunnel", - "vpnAddrs", vpnAddrs, "from", via, "laneIndex", result.PeerLaneIndex) - return - } - ls := base.lanes - - // The owner's lane index is bounded by its own advertised port count - // (lanes are one-per-routine and PortCount == routines on the owner). - laneIndex := result.PeerLaneIndex - if laneIndex >= uint32(ls.peerPortCount) || laneIndex > 256 { - hm.l.Debug("dropping lane handshake with out-of-range lane index", - "vpnAddrs", vpnAddrs, "from", via, "laneIndex", laneIndex) - return - } - - // Per-lane replay dedup against the existing lane with the same index. - stage0 := packet[header.Len:] - var existing *HostInfo - ls.Lock() - for _, h := range ls.peerLanes { - if h.laneIndex == uint16(laneIndex) { - existing = h - break - } - } - ls.Unlock() - if existing != nil { - if bytes.Equal(stage0, existing.HandshakePacket[handshakePacketStage0]) { - // Stage-0 retransmit: the peer is committed to the original - // response's ephemeral keys, resend it from the arrival socket. - if msg := existing.HandshakePacket[handshakePacketStage2]; msg != nil { - hm.sendHandshakeResponse(via, msg, existing, true) - } - return - } - if existing.lastHandshakeTime >= result.HandshakeTime { - existing.logger(hm.l).Debug("dropping stale lane handshake", - "laneIndex", laneIndex, "from", via) - return - } - // Newer handshake wins; the initiator only re-initiates after it - // declared the old lane dead. Silent local teardown. - hm.mainHostMap.DeleteHostInfo(existing) - } - - connState, err := newConnectionStateFromResult(result) - if err != nil { - hm.l.Error("Discarding lane handshake with an invalid message index", - "error", err, "vpnAddrs", vpnAddrs, "laneIndex", laneIndex) - return - } - - hostinfo := &HostInfo{ - ConnectionState: connState, - localIndexId: result.LocalIndex, - remoteIndexId: result.RemoteIndex, - vpnAddrs: vpnAddrs, - HandshakePacket: make(map[uint8][]byte, 0), - lastHandshakeTime: result.HandshakeTime, - relayState: RelayState{ - relays: nil, - relayForByAddr: map[netip.Addr]*Relay{}, - relayForByIdx: map[uint32]*Relay{}, - }, - // Private RemoteList: lane roaming must not touch the shared - // lighthouse-learned cache. - remotes: NewRemoteList(vpnAddrs, nil), - sockIdx: via.SockIdx, - laneIndex: uint16(laneIndex), - laneOwned: false, - parent: base, - } - - // packet aliases the listener's incoming buffer, so this copy must stay. - hostinfo.HandshakePacket[handshakePacketStage0] = make([]byte, len(stage0)) - copy(hostinfo.HandshakePacket[handshakePacketStage0], stage0) - if response != nil { - hostinfo.HandshakePacket[handshakePacketStage2] = response - } - hostinfo.SetRemote(via.UdpAddr) - hostinfo.buildNetworks(f.myVpnNetworksTable, result.RemoteCert.Certificate) - - // Index-collision checks, mirroring CheckAndComplete minus the Hosts - // dedup (lanes are keyed by (base, laneIndex), not by address). - hm.mainHostMap.Lock() - hm.Lock() - if existingIndex, found := hm.mainHostMap.Indexes[hostinfo.localIndexId]; found && existingIndex != hostinfo { - hm.Unlock() - hm.mainHostMap.Unlock() - hostinfo.logger(hm.l).Error("Failed to add lane due to localIndex collision", "laneIndex", laneIndex) - return - } - if existingPendingIndex, found := hm.indexes[hostinfo.localIndexId]; found && existingPendingIndex.hostinfo != hostinfo { - hm.Unlock() - hm.mainHostMap.Unlock() - hostinfo.logger(hm.l).Error("Failed to add lane due to pending localIndex collision", "laneIndex", laneIndex) - return - } - if existingRemoteIndex, found := hm.mainHostMap.RemoteIndexes[hostinfo.remoteIndexId]; found && existingRemoteIndex != nil { - hostinfo.logger(hm.l).Info("New lane shadows existing host remoteIndex", - "collision", existingRemoteIndex.vpnAddrs, - ) - } - hm.mainHostMap.unlockedAddLane(hostinfo, f) - hm.Unlock() - hm.mainHostMap.Unlock() - - ls.Lock() - ls.peerLanes = append(ls.peerLanes, hostinfo) - ls.Unlock() - - hostinfo.logger(hm.l).Info("Lane handshake received", - "laneIndex", laneIndex, - "from", via, - "initiatorIndex", result.RemoteIndex, - "responderIndex", result.LocalIndex, - ) - - hm.sendHandshakeResponse(via, response, hostinfo, false) -} - // continueHandshake feeds an incoming packet to an existing pending handshake Machine. func (hm *HandshakeManager) continueHandshake(via ViaSender, hh *HandshakeHostInfo, packet []byte) { f := hm.f @@ -1234,14 +885,6 @@ func (hm *HandshakeManager) continueHandshake(via ViaSender, hh *HandshakeHostIn } hostinfo := hh.hostinfo - if hostinfo.isLane() && via.IsRelayed { - // Lane handshakes are direct-only; a relayed continuation is a stray - // or a protocol violation. Drop without failing the machine so the - // direct stage-2 can still land. - f.l.Debug("dropping relayed lane handshake continuation", - "vpnAddrs", hostinfo.vpnAddrs, "from", via) - return - } if !via.IsRelayed { if !f.lightHouse.GetRemoteAllowList().AllowAll(hostinfo.vpnAddrs, via.UdpAddr.Addr()) { f.l.Debug("lighthouse.remote_allow_list denied incoming handshake", @@ -1254,7 +897,7 @@ func (hm *HandshakeManager) continueHandshake(via ViaSender, hh *HandshakeHostIn if machine == nil { f.l.Error("No handshake machine available for continuation", "vpnAddrs", hostinfo.vpnAddrs, "from", via) - hm.deletePendingHostInfo(hostinfo) + hm.DeleteHostInfo(hostinfo) return } @@ -1264,7 +907,7 @@ func (hm *HandshakeManager) continueHandshake(via ViaSender, hh *HandshakeHostIn if machine.Failed() { f.l.Warn("Failed to process handshake packet, abandoning", "vpnAddrs", hostinfo.vpnAddrs, "from", via, "error", err) - hm.deletePendingHostInfo(hostinfo) + hm.DeleteHostInfo(hostinfo) } else { f.l.Debug("Failed to process handshake packet", "vpnAddrs", hostinfo.vpnAddrs, "from", via, "error", err) @@ -1293,7 +936,7 @@ func (hm *HandshakeManager) continueHandshake(via ViaSender, hh *HandshakeHostIn if remoteCert == nil { f.l.Error("Handshake completed without peer certificate", "vpnAddrs", hostinfo.vpnAddrs, "from", via) - hm.deletePendingHostInfo(hostinfo) + hm.DeleteHostInfo(hostinfo) return } @@ -1327,7 +970,7 @@ func (hm *HandshakeManager) continueHandshake(via ViaSender, hh *HandshakeHostIn "issuer", remoteCert.Certificate.Issuer(), "handshake", m{"stage": uint64(machine.MessageIndex()), "style": header.SubTypeName(header.Handshake, machine.Subtype())}, ) - hm.deletePendingHostInfo(hostinfo) + hm.DeleteHostInfo(hostinfo) return } vpnAddrs[i] = network.Addr() @@ -1351,15 +994,6 @@ func (hm *HandshakeManager) continueHandshake(via ViaSender, hh *HandshakeHostIn "handshake", m{"stage": uint64(machine.MessageIndex()), "style": header.SubTypeName(header.Handshake, machine.Subtype())}, ) - if hostinfo.isLane() { - // No restart dance for lanes: close what the peer completed, - // release the slot with backoff, and let ensureLanes retry. - hm.deletePendingHostInfo(hostinfo) - hostinfo.vpnAddrs = vpnAddrs - f.sendCloseTunnel(hostinfo) - return - } - hm.DeleteHostInfo(hostinfo) hm.StartHandshake(hostinfo.vpnAddrs[0], func(newHH *HandshakeHostInfo) { newHH.hostinfo.remotes = hostinfo.remotes @@ -1394,31 +1028,8 @@ func (hm *HandshakeManager) continueHandshake(via ViaSender, hh *HandshakeHostIn hostinfo.vpnAddrs = vpnAddrs hostinfo.buildNetworks(f.myVpnNetworksTable, remoteCert.Certificate) - if hostinfo.isLane() { - hm.completeLane(hostinfo, f) - - // Publish to the data plane only after the lane is registered and its - // ConnectionState is fully populated; a routine that Loads non-nil - // must always see a usable tunnel. - ls := hostinfo.parent.lanes - i := int(hostinfo.laneIndex) - ls.Lock() - if i < len(ls.txPending) { - ls.txPending[i] = false - ls.txFails[i] = 0 - } - ls.Unlock() - if i < len(ls.txLanes) { - ls.txLanes[i].Store(hostinfo) - } - - f.metricHandshakes.Update(duration) - return - } - - hm.maybeAllocLaneState(hostinfo, result) + hm.maybeAllocLanes(hostinfo, result) hm.Complete(hostinfo, f) - hm.EnsureLanes(hostinfo) if len(hh.packetStore) > 0 { if f.l.Enabled(context.Background(), slog.LevelDebug) { diff --git a/header/header.go b/header/header.go index d0848c55..19cc7011 100644 --- a/header/header.go +++ b/header/header.go @@ -12,6 +12,9 @@ import ( // |-----------------------------------------------------------------------| // | Version (uint4) | Type (uint4) | Subtype (uint8) | Reserved (uint16) | 32 // |-----------------------------------------------------------------------| +// The low 8 bits of Reserved carry the multiport lane index (0 is the base +// tunnel, which is what every non-multiport sender emits). The high 8 bits +// remain reserved and are always sent as zero. // | Remote index (uint32) | 64 // |-----------------------------------------------------------------------| // | Message counter | 96 @@ -57,8 +60,18 @@ const ( const ( TestRequest MessageSubType = 0 TestReply MessageSubType = 1 + // LaneProbe is sent on a multiport lane to prove the lane's 5-tuple is + // usable; LaneProbeAck answers it on the base tunnel. + LaneProbe MessageSubType = 2 + LaneProbeAck MessageSubType = 3 ) +// MaxLane is the largest lane index the header can carry. +const MaxLane = 0xff + +// laneMask covers the bits of Reserved that hold the lane index. +const laneMask uint16 = 0x00ff + const ( HandshakeIXPSK0 MessageSubType = 0 HandshakeXXPSK0 MessageSubType = 1 @@ -67,8 +80,10 @@ const ( var ErrHeaderTooShort = errors.New("header is too short") var subTypeTestMap = map[MessageSubType]string{ - TestRequest: "testRequest", - TestReply: "testReply", + TestRequest: "testRequest", + TestReply: "testReply", + LaneProbe: "laneProbe", + LaneProbeAck: "laneProbeAck", } var subTypeNoneMap = map[MessageSubType]string{0: "none"} @@ -100,10 +115,16 @@ type H struct { // Encode uses the provided byte array to encode the provided header values into. // Byte array must be capped higher than HeaderLen or this will panic func Encode(b []byte, v uint8, t MessageType, st MessageSubType, ri uint32, c uint64) []byte { + return EncodeLane(b, v, t, st, ri, c, 0) +} + +// EncodeLane is Encode with an explicit multiport lane index, which is carried +// in the low 8 bits of Reserved. +func EncodeLane(b []byte, v uint8, t MessageType, st MessageSubType, ri uint32, c uint64, lane uint8) []byte { b = b[:Len] b[0] = v<<4 | byte(t&0x0f) b[1] = byte(st) - binary.BigEndian.PutUint16(b[2:4], 0) + binary.BigEndian.PutUint16(b[2:4], uint16(lane)) binary.BigEndian.PutUint32(b[4:8], ri) binary.BigEndian.PutUint64(b[8:16], c) return b @@ -136,7 +157,13 @@ func (h *H) Encode(b []byte) ([]byte, error) { return nil, errors.New("nil header") } - return Encode(b, h.Version, h.Type, h.Subtype, h.RemoteIndex, h.MessageCounter), nil + return EncodeLane(b, h.Version, h.Type, h.Subtype, h.RemoteIndex, h.MessageCounter, h.Lane()), nil +} + +// Lane returns the multiport lane index carried in Reserved. Lane 0 is the base +// tunnel, which is what any sender that does not know about lanes will report. +func (h *H) Lane() uint8 { + return uint8(h.Reserved & laneMask) } // Parse is a helper function to parses given bytes into new Header struct @@ -196,7 +223,7 @@ func IsValidSubType(t MessageType, s MessageSubType) bool { case Handshake: return s == HandshakeIXPSK0 case Test: - return s == TestReply || s == TestRequest + return s == TestReply || s == TestRequest || s == LaneProbe || s == LaneProbeAck case Control, CloseTunnel, RecvError, LightHouse: return s == 0 default: diff --git a/header/header_test.go b/header/header_test.go index 63c5b1a5..85781449 100644 --- a/header/header_test.go +++ b/header/header_test.go @@ -52,6 +52,28 @@ func TestParse(t *testing.T) { } } +func TestEncodeLane(t *testing.T) { + b := EncodeLane(make([]byte, Len), Version, Message, MessageNone, 10, 9, 3) + assert.Equal(t, []byte{0x11, 0x0, 0x0, 0x3}, b[:4]) + + h := &H{} + require.NoError(t, h.Parse(b)) + assert.Equal(t, uint16(3), h.Reserved) + assert.Equal(t, uint8(3), h.Lane()) + + // Encode is EncodeLane on the base tunnel, and the H method round trips the lane. + assert.Equal(t, + EncodeLane(make([]byte, Len), Version, Message, MessageNone, 10, 9, 0), + Encode(make([]byte, Len), Version, Message, MessageNone, 10, 9)) + + rt, err := h.Encode(make([]byte, Len)) + require.NoError(t, err) + assert.Equal(t, b, rt) + + // Only the low 8 bits of Reserved are the lane. + assert.Equal(t, uint8(0x2a), (&H{Reserved: 0xff2a}).Lane()) +} + func TestTypeName(t *testing.T) { assert.Equal(t, "test", TypeName(Test)) assert.Equal(t, "test", (&H{Type: Test}).TypeName()) @@ -127,7 +149,9 @@ func TestIsValidSubType(t *testing.T) { assert.True(t, IsValidSubType(Test, TestRequest)) assert.True(t, IsValidSubType(Test, TestReply)) - assert.False(t, IsValidSubType(Test, 2)) + assert.True(t, IsValidSubType(Test, LaneProbe)) + assert.True(t, IsValidSubType(Test, LaneProbeAck)) + assert.False(t, IsValidSubType(Test, 4)) // These types only ever carry subtype 0. for _, mt := range []MessageType{Control, CloseTunnel, RecvError, LightHouse} { diff --git a/hostmap.go b/hostmap.go index dae94cd6..ada2e375 100644 --- a/hostmap.go +++ b/hostmap.go @@ -5,7 +5,6 @@ import ( "encoding/json" "errors" "fmt" - "hash/fnv" "log/slog" "net" "net/netip" @@ -280,208 +279,10 @@ type HostInfo struct { // This should only be used by the ConnectionManagers ticker routine. lastUsed time.Time - // sockIdx is the index into Interface.writers of the socket every packet - // on this tunnel egresses from (and, for lanes, arrives on). 0 for base - // and vanilla tunnels — the zero value preserves stock behavior. - sockIdx int - - // laneIndex is the owner's lane number for a lane tunnel; 0 for base. - laneIndex uint16 - - // laneOwned is true when we initiated this lane (it carries our TX data). - laneOwned bool - - // parent points at the base tunnel a lane hangs off of; nil for base and - // vanilla tunnels. Set before the lane is registered in hostmap.Indexes. - parent *HostInfo - - // lanes is allocated on a base tunnel when multiport is enabled and the - // peer advertised lane support; nil otherwise. - lanes *laneState -} - -// isLane reports whether this HostInfo is a lane tunnel rather than a base -// (or vanilla) tunnel. -func (i *HostInfo) isLane() bool { - return i.parent != nil -} - -// laneState hangs off a base HostInfo and tracks the multiport lane tunnels -// associated with it. txLanes is read lock-free on the TX hot path; the Mutex -// guards everything else. -type laneState struct { - sync.Mutex - - // peerPortCount/peerBasePort are the peer's advert from the base - // handshake; portOffset is the per-pair rotation from lanePortOffset. - // Lane i targets peerBasePort + ((i + portOffset) % peerPortCount). - peerPortCount uint16 - peerBasePort uint16 - portOffset uint16 - - // txLanes[i] is our established, initiator-owned lane for routine i, or - // nil. Index 0 is always nil — the base tunnel is lane 0. A pointer is - // only Stored once the lane's ConnectionState is fully populated, so a - // data-plane routine that Loads non-nil always sees a usable tunnel. - txLanes []atomic.Pointer[HostInfo] - - // txDemand[i] is set by the data plane when a routine riding slot i has - // traffic for this peer and no established lane, and consumed when - // EnsureLanes claims the slot. Lanes exist only where traffic asked for - // one: a peer we exchange a trickle with never costs more than the base - // tunnel, no matter how many routines are configured. Written without - // the Mutex. - txDemand []atomic.Bool - - // Under Mutex: per-slot handshake-in-flight flag, consecutive failure - // count, and earliest next attempt, driving ensureLanes' backoff. - txPending []bool - txFails []uint8 - txRetryAt []time.Time - - // Under Mutex: responder-side records of peer-owned lanes, capped by - // same-laneIndex replacement. - peerLanes []*HostInfo -} - -func newLaneState(laneCount int, peerPortCount, peerBasePort, portOffset uint16) *laneState { - return &laneState{ - peerPortCount: peerPortCount, - peerBasePort: peerBasePort, - portOffset: portOffset, - txLanes: make([]atomic.Pointer[HostInfo], laneCount), - txDemand: make([]atomic.Bool, laneCount), - txPending: make([]bool, laneCount), - txFails: make([]uint8, laneCount), - txRetryAt: make([]time.Time, laneCount), - } -} - -// noteLaneDemand records that a data-plane routine has traffic for lane slot -// i but found the slot empty. Called from the TX hot path on every packet -// that misses, so the store is load-guarded: once the flag is up, further -// misses are plain reads and cannot ping-pong a cache line shared with the -// neighbouring slots' flags. Slot 0 is the base tunnel and never a lane. -func (ls *laneState) noteLaneDemand(i int) { - if i <= 0 || i >= len(ls.txDemand) { - return - } - if !ls.txDemand[i].Load() { - ls.txDemand[i].Store(true) - } -} - -// takeLaneDemand consumes slot i's demand flag, reporting whether the data -// plane had asked for the lane since the last time we looked. -func (ls *laneState) takeLaneDemand(i int) bool { - if i <= 0 || i >= len(ls.txDemand) { - return false - } - return ls.txDemand[i].Swap(false) -} - -// laneTargetPort returns the peer port that owned lane i handshakes to and -// egresses toward. The caller must ensure peerPortCount != 0. -func (ls *laneState) laneTargetPort(i int) uint16 { - return ls.peerBasePort + uint16((i+int(ls.portOffset))%int(ls.peerPortCount)) -} - -// lanePortOffset returns the rotation applied to this pair's lane target -// ports, in [0, peerPortCount). Without it every low-routine peer would aim -// its few lanes at a big peer's first few ports, concentrating the big -// peer's receive work on a couple of sockets; the hash spreads pairs across -// the whole advertised range. -// -// Both sides hash the same sorted vpn-address pair and the higher address -// negates the result, so when port counts match the two sides' rotations -// cancel: our lane i's 4-tuple is still the reverse of a peer-owned lane's, -// and each outbound lane handshake opens the conntrack entry its partner -// arrives through. (The one lane a nonzero rotation lands on the peer's base -// port has no partner lane; behind a port-restricted NAT it may not form and -// its routine rides the base tunnel — the standard lane fallback.) -func lanePortOffset(myAddr, peerAddr netip.Addr, peerPortCount uint16) uint16 { - if peerPortCount == 0 { - return 0 - } - lo, hi := myAddr, peerAddr - if hi.Less(lo) { - lo, hi = hi, lo - } - h := fnv.New32a() - b := lo.As16() - h.Write(b[:]) - b = hi.As16() - h.Write(b[:]) - o := uint16(h.Sum32() % uint32(peerPortCount)) - if myAddr == hi { - o = (peerPortCount - o) % peerPortCount - } - return o -} - -const ( - laneRetryBase = 5 * time.Second - laneRetryMax = 60 * time.Second -) - -// noteLaneFailure marks lane slot i as empty and pushes the next attempt out -// with exponential backoff. Called when an owned lane dies or its handshake -// times out. -func (ls *laneState) noteLaneFailure(i int) { - if i < 0 || i >= len(ls.txPending) { - return - } - ls.Lock() - ls.txPending[i] = false - if ls.txFails[i] < 200 { // just avoid wrapping; the delay caps far earlier - ls.txFails[i]++ - } - d := laneRetryBase << min(ls.txFails[i], 4) - if d > laneRetryMax { - d = laneRetryMax - } - ls.txRetryAt[i] = time.Now().Add(d) - ls.Unlock() -} - -// noteOwnedLaneDeath detaches an established owned lane from its slot -// (identity-checked, so a raced re-establishment is never clobbered) and -// applies failure backoff. -func (ls *laneState) noteOwnedLaneDeath(lane *HostInfo) { - i := int(lane.laneIndex) - if i >= len(ls.txLanes) { - return - } - ls.txLanes[i].CompareAndSwap(lane, nil) - ls.noteLaneFailure(i) -} - -// removePeerLane drops a responder-side lane record by identity. -func (ls *laneState) removePeerLane(lane *HostInfo) { - ls.Lock() - for n, h := range ls.peerLanes { - if h == lane { - ls.peerLanes = append(ls.peerLanes[:n], ls.peerLanes[n+1:]...) - break - } - } - ls.Unlock() -} - -// snapshotLanes returns every lane hostinfo currently attached, used by the -// base-delete cascade. Taken under the lock and returned as a copy so the -// caller can delete without holding it. -func (ls *laneState) snapshotLanes() []*HostInfo { - ls.Lock() - defer ls.Unlock() - out := make([]*HostInfo, 0, len(ls.txLanes)+len(ls.peerLanes)) - for n := range ls.txLanes { - if h := ls.txLanes[n].Load(); h != nil { - out = append(out, h) - } - } - out = append(out, ls.peerLanes...) - return out + // lanes holds this tunnel's multiport lane sessions. Allocated when the + // handshake completes if both sides advertised multiport, nil otherwise. + // Immutable once the hostinfo is published to the data plane. + lanes *laneSet } type ViaSender struct { @@ -655,12 +456,6 @@ func (hm *HostMap) MakePrimary(hostinfo *HostInfo) { // unlockedMakePrimary reports whether hostinfo is (now) the primary for each of its addresses, // false only when it is no longer in the hostmap at all. func (hm *HostMap) unlockedMakePrimary(hostinfo *HostInfo) bool { - // A lane must never become a Hosts primary: it would start carrying all - // traffic for the peer and become a relay candidate. - if hostinfo.isLane() { - return false - } - // A hostinfo that is no longer in the hostmap must not be re-inserted here. Callers can race // tunnel teardown, deciding to promote under the read lock and only taking the write lock // after a delete fully unlinked the hostinfo (connection manager swapPrimary, AddRelay). Every @@ -689,18 +484,8 @@ func (hm *HostMap) unlockedMakePrimary(hostinfo *HostInfo) bool { // any tunnel to the peer), which the caller uses to decide whether to clear learned lighthouse // state and disestablish relays. func (hm *HostMap) unlockedDeleteHostInfo(hostinfo *HostInfo) bool { - if hostinfo.isLane() { - return hm.unlockedDeleteLane(hostinfo) - } - - // A dying base takes its lanes with it. The peer converges symmetrically - // when it processes the base's CloseTunnel, so lanes need no signaling of - // their own. Depth-1 recursion: lanes have no children. - if hostinfo.lanes != nil { - for _, lane := range hostinfo.lanes.snapshotLanes() { - hm.unlockedDeleteLane(lane) - } - } + // Lane sessions hang off this hostinfo, so deleting it takes them with it + // and there is nothing extra to unwind here. // Remove this hostinfo from each of its address lists. The lists are independent, so a // sibling is never promoted to an address it does not own and no other list is touched. @@ -778,35 +563,6 @@ func (hm *HostMap) QueryIndexCached(index uint32, cache map[uint32]*HostInfo) *H return out } -// unlockedDeleteLane removes a lane tunnel from the index maps and detaches it -// from its base. Lanes never live in Hosts and their death never means "no -// tunnel to the peer", so the return is always false (the lighthouse cache and -// relays stay untouched). Idempotent: every step is identity-checked. -func (hm *HostMap) unlockedDeleteLane(lane *HostInfo) bool { - if ls := lane.parent.lanes; ls != nil { - if lane.laneOwned { - ls.noteOwnedLaneDeath(lane) - } else { - ls.removePeerLane(lane) - } - } - - if hostinfo2, ok := hm.RemoteIndexes[lane.remoteIndexId]; ok && hostinfo2 == lane { - delete(hm.RemoteIndexes, lane.remoteIndexId) - } - if hostinfo2, ok := hm.Indexes[lane.localIndexId]; ok && hostinfo2 == lane { - delete(hm.Indexes, lane.localIndexId) - } - - if hm.l.Enabled(context.Background(), slog.LevelDebug) { - hm.l.Debug("Hostmap lane deleted", - "hostMap", m{"vpnAddrs": lane.vpnAddrs, "laneIndex": lane.laneIndex, - "indexNumber": lane.localIndexId, "remoteIndexNumber": lane.remoteIndexId}, - ) - } - return false -} - func (hm *HostMap) QueryIndex(index uint32) *HostInfo { hm.RLock() if h, ok := hm.Indexes[index]; ok { @@ -935,27 +691,6 @@ func (hm *HostMap) unlockedAddHostInfo(hostinfo *HostInfo, f *Interface) { } } -// unlockedAddLane registers a lane tunnel in the index maps (RX demux and -// recv_error need it there) without touching Hosts: lanes are never primary, -// never dns-visible, and never subject to the MaxHostInfosPerVpnIp eviction. -// The connection manager still tracks it for keepalive/death. -func (hm *HostMap) unlockedAddLane(lane *HostInfo, f *Interface) { - hm.Indexes[lane.localIndexId] = lane - hm.RemoteIndexes[lane.remoteIndexId] = lane - - lane.markOutOnly() - if f.connectionManager != nil { // f.connectionManager is only nil in some unit tests - f.connectionManager.trafficTimer.Add(lane.localIndexId, f.connectionManager.checkInterval) - } - - if hm.l.Enabled(context.Background(), slog.LevelDebug) { - hm.l.Debug("Hostmap lane added", - "hostMap", m{"vpnAddrs": lane.vpnAddrs, "laneIndex": lane.laneIndex, - "indexNumber": lane.localIndexId, "remoteIndexNumber": lane.remoteIndexId}, - ) - } -} - func (hm *HostMap) unlockedInnerAddHostInfo(vpnAddr netip.Addr, hostinfo *HostInfo, f *Interface) { existing, ok := hm.Hosts[vpnAddr] if !ok { diff --git a/inside.go b/inside.go index 90dc1349..83337db5 100644 --- a/inside.go +++ b/inside.go @@ -122,13 +122,13 @@ func (f *Interface) consumeInsidePacket(pkt tio.Packet, fwPacket *firewall.Parse } } -func (f *Interface) sendInsideEncrypt(hostinfo *HostInfo, ci *ConnectionState, seg, scratch, nb []byte) []byte { +func (f *Interface) sendInsideEncrypt(hostinfo *HostInfo, ci *ConnectionState, lane uint8, seg, scratch, nb []byte) []byte { if noiseutil.EncryptLockNeeded { ci.writeLock.Lock() } c := ci.messageCounter.Add(1) - out := header.Encode(scratch, header.Version, header.Message, 0, hostinfo.remoteIndexId, c) + out := header.EncodeLane(scratch, header.Version, header.Message, 0, hostinfo.remoteIndexId, c, lane) out, encErr := ci.eKey.EncryptDanger(out, out, seg, c, nb) if noiseutil.EncryptLockNeeded { @@ -154,9 +154,8 @@ func (f *Interface) sendInsideEncrypt(hostinfo *HostInfo, ci *ConnectionState, s // scratch arena: SegmentSuperpacket builds each segment's plaintext in // segScratch[:segLen] in turn, and we encrypt directly into a fresh SendBatch slot. // -// hostinfo is always the base tunnel (the hostmap resolves by vpn address); -// when routine q has an established lane to this peer, the direct path swaps -// to the lane's session and socket below. Relay and base traffic stays on +// When routine q has a usable multiport lane to this peer, the direct path +// swaps to the lane's session and socket below. Relay and base traffic stays on // tx.base (socket 0). func (f *Interface) sendInsideMessage(hostinfo *HostInfo, pkt tio.Packet, nb []byte, tx *txQueue) { ci := hostinfo.ConnectionState @@ -208,7 +207,7 @@ func (f *Interface) sendInsideMessage(hostinfo *HostInfo, pkt tio.Packet, nb []b //relay header + header + plaintext + AEAD tag (16 bytes for both AES-GCM and ChaCha20-Poly1305) + relay tag scratch := sendBatch.Reserve(header.Len + header.Len + len(seg) + 16 + 16) - innerPacket := f.sendInsideEncrypt(hostinfo, ci, seg, scratch[header.Len:], nb) + innerPacket := f.sendInsideEncrypt(hostinfo, ci, 0, seg, scratch[header.Len:], nb) if innerPacket == nil { return nil } @@ -229,35 +228,28 @@ func (f *Interface) sendInsideMessage(hostinfo *HostInfo, pkt tio.Packet, nb []b return } - // Direct path: prefer this routine's lane tunnel when it is established. - // The pointer is only published once the lane's ConnectionState is fully - // populated, so a non-nil Load is always usable. On lane death the slot - // CAS-clears and traffic falls back to the base tunnel instantly. + // Direct path: prefer this routine's multiport lane once it is proven + // usable. txLane hands back the lane's session and destination together, so + // there is no window where one is set and the other is not, and a demotion + // drops us back onto the base tunnel on the very next packet. // - // A miss is also how lanes get built in the first place: flagging demand - // here is the only thing that asks the handshake manager for this slot, - // so we pay for a lane exactly where real traffic wanted one. The - // connection manager's next tick on this tunnel picks the flag up, which - // bounds establishment by one check interval — until then the traffic - // rides the base tunnel, the same fallback a dead lane uses. - if ls := hostinfo.lanes; ls != nil && tx.laneSlot < len(ls.txLanes) { - if lane := ls.txLanes[tx.laneSlot].Load(); lane != nil { - if lci := lane.ConnectionState; lci != nil && lci.eKey != nil { - hostinfo = lane - ci = lci - remote = lane.GetRemote() - sendBatch = tx.lane - } - } else { - ls.noteLaneDemand(tx.laneSlot) - } + // A miss is also how lanes get probed in the first place: txLane raises + // demand, which the connection manager's next tick on this tunnel picks up. + // Until the lane is up the traffic rides the base tunnel, the same fallback + // a demoted lane uses. + lane := uint8(0) + if lci, laneRemote := hostinfo.lanes.txLane(tx.laneSlot); lci != nil { + lane = uint8(tx.laneSlot) + ci = lci + remote = laneRemote + sendBatch = tx.lane } err := tio.SegmentSuperpacket(pkt, func(seg []byte) error { // header + plaintext + AEAD tag (16 bytes for both AES-GCM and ChaCha20-Poly1305) scratch := sendBatch.Reserve(header.Len + len(seg) + 16) - out := f.sendInsideEncrypt(hostinfo, ci, seg, scratch, nb) + out := f.sendInsideEncrypt(hostinfo, ci, lane, seg, scratch, nb) if out == nil { return nil } @@ -546,8 +538,7 @@ func (f *Interface) SendVia(via *HostInfo, relay *Relay, ad, nb, out []byte, noc return } - // Relay carriers are base tunnels (sockIdx 0); indexing through the carrier keeps the invariant explicit. - err = f.writers[f.egressSock(via, q)].WriteTo(toSend, via.GetRemote()) + err = f.writers[f.egressSock(q)].WriteTo(toSend, via.GetRemote()) if err != nil { via.logger(f.l).Info("Failed to WriteTo in sendVia", "error", err) } @@ -555,15 +546,15 @@ func (f *Interface) SendVia(via *HostInfo, relay *Relay, ad, nb, out []byte, noc // egressSock picks the socket a tunnel packet leaves from. // -// With multiport the tunnel's own sockIdx is authoritative: a lane must egress the lane's 4-tuple so the peer's -// spoof/roam checks accept keepalives, close packets and rejects and the NAT entry stays warm, and a base or relay -// carrier must egress socket 0 so a vanilla peer never sees per-routine source ports. +// With multiport, everything that is not lane data plane egresses socket 0: handshakes, keepalives, close packets, +// rejects and relay carriers all belong to the base tunnel's 4-tuple, which is the only one a peer's spoof/roam checks +// and a vanilla peer's expectations know about. Lane data is sent directly through writers[lane] and never comes here. // // Without multiport every socket shares one port under SO_REUSEPORT, so the source address is identical either way and // we keep q, the socket the triggering packet arrived on, to avoid contending on socket 0's fd. -func (f *Interface) egressSock(hostinfo *HostInfo, q int) int { +func (f *Interface) egressSock(q int) int { if f.multiport { - return hostinfo.sockIdx + return 0 } return q } @@ -572,7 +563,7 @@ func (f *Interface) sendNoMetrics(t header.MessageType, st header.MessageSubType if ci.eKey == nil { return } - q = f.egressSock(hostinfo, q) + q = f.egressSock(q) useRelay := !remote.IsValid() && !hostinfo.GetRemote().IsValid() fullOut := out @@ -604,10 +595,7 @@ func (f *Interface) sendNoMetrics(t header.MessageType, st header.MessageSubType // 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. - // - // isLane is checked last on purpose: a lane still needs its Out() traffic recorded for the liveness decision, it - // just shouldn't issue the lighthouse query. The base tunnel issues the one query for the peer. - if t != header.CloseTunnel && f.connectionManager.Out(hostinfo) && !hostinfo.isLane() { + if t != header.CloseTunnel && 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", @@ -647,13 +635,6 @@ func (f *Interface) sendNoMetrics(t header.MessageType, st header.MessageSubType ) } } else { - if hostinfo.isLane() { - // A lane always has a valid remote (set from its own handshake); - // reaching here means the lane is broken, and lane ciphertext must - // never ride a relay (relays are base-tunnel-only). - hostinfo.logger(f.l).Error("Dropping lane packet with no valid remote") - return - } // Try to send via a relay for _, relayIP := range hostinfo.relayState.CopyRelayIps() { relayHostInfo, relay, err := f.hostMap.QueryVpnAddrsRelayFor(hostinfo.vpnAddrs, relayIP) diff --git a/interface.go b/interface.go index a386befc..e20315e0 100644 --- a/interface.go +++ b/interface.go @@ -452,10 +452,12 @@ func (f *Interface) pinThisThread(i int) { // txQueue is the per-routine TX state owned by one listenIn goroutine. // laneSlot is the lane this routine's traffic rides (laneSlotFor); lane is -// bound to that slot's socket and carries lane-tunnel data; base is bound to -// socket 0 and carries base-tunnel and relay data, which must keep the base -// source port (a vanilla peer would otherwise see per-routine source ports -// and roam-thrash). The two alias when multiport is off or laneSlot is 0. +// bound to that slot's socket and carries traffic encrypted with that lane's +// session; base is bound to socket 0 and carries base-session and relay data, +// which must keep the base source port (a vanilla peer would otherwise see +// per-routine source ports and roam-thrash). A routine uses base whenever its +// lane is down, so both batches stay live for the life of the routine. The two +// alias when multiport is off or laneSlot is 0. // Concurrent sendmmsg on a shared fd (socket 0, or a lane socket shared by // overflow routines) is safe: a flow is pinned to one routine by tun // steering, so per-flow wire order still holds. @@ -473,8 +475,8 @@ func (tx *txQueue) full() bool { } // flush drains base before lane so that when a flow moves from the base -// tunnel onto a freshly established lane mid-window, its packets still leave -// this host in encryption order. +// session onto a freshly promoted lane mid-window, its packets still leave this +// host in encryption order. func (tx *txQueue) flush(f *Interface) { if tx.base != tx.lane { f.flushSendBatch(tx.base, 0) @@ -485,9 +487,10 @@ func (tx *txQueue) flush(f *Interface) { // laneSlotFor maps a routine index to the lane its traffic rides. When // multiport.lanes is below routines, overflow routines share the configured // lanes round-robin instead of all falling back onto the base tunnel's -// single underlay flow. Sharers use the lane's own socket and 4-tuple, so -// this is vanilla-style same-flow sharing: no cross-path replay skew, and -// per-flow ordering still holds (a flow stays pinned to one routine). +// single underlay flow. Sharers use the lane's own socket, 4-tuple and +// session, so this is vanilla-style same-flow sharing: no cross-path replay +// skew, and per-flow ordering still holds (a flow stays pinned to one +// routine). func (f *Interface) laneSlotFor(i int) int { if f.multiport && f.laneCount > 0 { return i % f.laneCount diff --git a/lanes.go b/lanes.go new file mode 100644 index 00000000..a4ce547f --- /dev/null +++ b/lanes.go @@ -0,0 +1,462 @@ +package nebula + +import ( + "context" + "hash/fnv" + "log/slog" + "net/netip" + "sync" + "sync/atomic" + "time" + + "github.com/slackhq/nebula/handshake" + "github.com/slackhq/nebula/header" + "github.com/slackhq/nebula/noiseutil" +) + +// Multiport lanes give each data-plane routine its own underlay 5-tuple, so one +// tunnel's traffic spreads over ECMP paths, NIC receive queues and per-flow +// policers instead of funnelling through a single flow. +// +// A lane is not a second tunnel: it is an extra session on the same HostInfo. +// Noise leaves us with A.eKey == B.dKey, so both sides expand the same two keys +// with the same per-lane label and land on a matched pair without exchanging +// anything. A lane therefore costs no handshake, has no half-established state, +// and dies exactly when its base tunnel does. Which lane a packet belongs to +// travels in the nebula header, inside the AEAD's associated data. +// +// Lane 0 is the base tunnel itself: HostInfo.ConnectionState, socket 0, and the +// peer's real remote address. Lane s > 0 egresses writers[s] (bound to +// listen.port+s) toward the peer's advertised port range. Receiving on a lane +// needs no permission — the session exists as soon as the base handshake +// completes — but sending on one needs proof the new 5-tuple actually works, +// since nothing else would notice a middlebox quietly dropping it. So a lane +// stays down until a probe on it is acked, and falls back to the base tunnel the +// moment it stops being acked. + +const ( + // laneKeyInfo is the HKDF label prefix for lane key expansion. Changing it + // means older builds derive different keys and drop our lane traffic; the + // base tunnel would keep working, so the failure would be a silent loss of + // lanes rather than of connectivity. + laneKeyInfo = "nebula multiport lane v1" + + // laneRetryBase and laneRetryMax bound the backoff between probes of a lane + // that will not come up, so a peer whose lane ports are firewalled costs one + // packet a minute rather than one per traffic tick. + laneRetryBase = 5 * time.Second + laneRetryMax = 60 * time.Second + + // laneMaxFails caps the failure counter; the backoff saturates well before. + laneMaxFails = 8 + + // laneProbeTimeout is how long a probe may go unacked before it counts as a + // failure. It is shorter than the connection manager's check interval on + // purpose: an outstanding probe is judged on the next tick either way, and a + // longer timeout would only delay that by a whole tick. + laneProbeTimeout = 2 * time.Second + + // laneKeepalive is how often a lane that is up re-proves its path. Traffic + // on a lane is not evidence the lane works — that is the whole reason lanes + // need probing — so a lane that silently breaks is only caught here. + laneKeepalive = 30 * time.Second +) + +// laneSet holds a peer's lane sessions and the state deciding which lanes may +// carry traffic. It is built when the base handshake completes and never +// resized, so the slices and their lengths are immutable; mu guards the fields +// under it, and txAddr/demand are atomics read by the data plane without it. +type laneSet struct { + // sessions[s] is lane s's session. sessions[0] is nil: lane 0 is the base + // tunnel's own ConnectionState. Immutable, so the RX path reads it with no + // synchronization at all. + sessions []*ConnectionState + + // txAddr[s] holds lane s's remote address while the lane is proven usable + // and nil otherwise. This single atomic is both the TX gate and the + // destination, so a routine that loads non-nil has everything it needs. + txAddr []atomic.Pointer[netip.AddrPort] + + // demand[s] is raised by the TX path when a routine riding lane s has + // traffic for this peer and the lane is down. Probing is demand-driven: a + // peer we exchange a trickle with never costs more than its base tunnel, no + // matter how many lanes are configured. + demand []atomic.Bool + + // txLanes is how many lanes we may send on — our lane count clamped to the + // ports the peer bound. Lanes from txLanes up exist only to receive, which + // is how a peer with more routines than us still spreads its own traffic. + // Immutable. + txLanes int + + mu sync.Mutex + + // peerPortCount and peerBasePort are the peer's advertised port range and + // portOffset is this pair's rotation within it: lane s targets + // peerBasePort + ((s + portOffset) % peerPortCount). + peerPortCount uint16 + peerBasePort uint16 + portOffset uint16 + + // peerAddr is the address the current lane targets were built from. The + // peer's lane ports have no derivable relationship to a new NAT mapping, so + // a roam invalidates every lane rather than moving it. + peerAddr netip.Addr + + // probe[s] is lane s's probe and backoff state. + probe []laneProbeState +} + +type laneProbeState struct { + // gen is the generation of the last probe sent, echoed in the ack so a late + // ack cannot promote a lane on the strength of a superseded probe. + gen uint8 + + // fails is the consecutive failure count driving retryAt. + fails uint8 + + // sentAt is when the outstanding probe went out, zero when none is pending. + sentAt time.Time + + // target is where the outstanding probe went, promoted to txAddr on ack. + target netip.AddrPort + + // lastAck is when the lane was last confirmed usable, driving the keepalive. + lastAck time.Time + + // retryAt is the earliest we may probe this lane again. + retryAt time.Time +} + +// newLaneSet derives the lane sessions for a freshly completed base handshake. +// It returns nil when the pair has no lane beyond the base tunnel, which is the +// normal answer for a peer running without multiport. +func newLaneSet(r *handshake.Result, myLanes int, myAddr, peerAddr netip.Addr) (*laneSet, error) { + // PeerPortCount and PeerBasePort are already bounded to uint16 by the + // handshake payload parser. A zero port count is a peer that did not + // advertise multiport at all, so there is no lane to be had in either + // direction. + peerPorts := uint16(r.PeerPortCount) + if peerPorts == 0 { + return nil, nil + } + + // Sessions have to cover both directions: we send on our lanes and receive + // on the peer's, and one derived session serves both ends of a lane index. + n := min(max(myLanes, int(r.PeerTxLanes)), header.MaxLane+1) + if n < 2 { + return nil, nil + } + + ls := &laneSet{ + sessions: make([]*ConnectionState, n), + txAddr: make([]atomic.Pointer[netip.AddrPort], n), + demand: make([]atomic.Bool, n), + probe: make([]laneProbeState, n), + txLanes: min(myLanes, int(peerPorts), n), + peerPortCount: peerPorts, + peerBasePort: uint16(r.PeerBasePort), + portOffset: lanePortOffset(myAddr, peerAddr, peerPorts), + } + + for s := 1; s < n; s++ { + cs, err := newLaneConnectionState(r, uint8(s)) + if err != nil { + return nil, err + } + ls.sessions[s] = cs + } + + return ls, nil +} + +// lanePortOffset returns the rotation applied to this pair's lane target ports, +// in [0, peerPortCount). Without it every low-routine peer would aim its few +// lanes at a big peer's first few ports, concentrating the big peer's receive +// work on a couple of sockets; the hash spreads pairs across the whole range. +// +// Both sides hash the same sorted vpn-address pair and the higher address +// negates the result, so when port counts match the two sides' rotations +// cancel: our lane s's 4-tuple stays the reverse of the peer's lane s, and each +// side's probe opens the conntrack entry the other's arrives through. (The one +// lane a nonzero rotation lands on the peer's base port has no partner lane; +// behind a port-restricted NAT it may never come up, and its routine rides the +// base tunnel — the standard lane fallback.) +func lanePortOffset(myAddr, peerAddr netip.Addr, peerPortCount uint16) uint16 { + if peerPortCount == 0 { + return 0 + } + lo, hi := myAddr, peerAddr + if hi.Less(lo) { + lo, hi = hi, lo + } + h := fnv.New32a() + b := lo.As16() + h.Write(b[:]) + b = hi.As16() + h.Write(b[:]) + o := uint16(h.Sum32() % uint32(peerPortCount)) + if myAddr == hi { + o = (peerPortCount - o) % peerPortCount + } + return o +} + +// laneSession returns lane s's session, or nil if this peer has no lane s. +func (i *HostInfo) laneSession(s uint8) *ConnectionState { + ls := i.lanes + if ls == nil || int(s) >= len(ls.sessions) { + return nil + } + return ls.sessions[s] +} + +// maxMessageCounter returns the highest counter across the base session and +// every lane session. Data rides the lanes, so the base counter alone would +// never reach the rehandshake or exhaustion thresholds and the lane keys would +// be used past their data-volume margin. Rolling the base tunnel replaces the +// lane keys with it, since lanes are derived from it. +func (i *HostInfo) maxMessageCounter() uint64 { + if i.ConnectionState == nil { + return 0 + } + c := i.ConnectionState.messageCounter.Load() + if ls := i.lanes; ls != nil { + for _, cs := range ls.sessions { + if cs == nil { + continue + } + if lc := cs.messageCounter.Load(); lc > c { + c = lc + } + } + } + return c +} + +// txLane returns the session and destination for lane s, or a nil session when +// the lane is down and the caller must use the base tunnel. A miss raises +// demand, which is the only thing that gets a lane probed, so we pay for a lane +// exactly where real traffic wanted one. +func (ls *laneSet) txLane(s int) (*ConnectionState, netip.AddrPort) { + if ls == nil || s <= 0 || s >= ls.txLanes { + return nil, netip.AddrPort{} + } + + if addr := ls.txAddr[s].Load(); addr != nil { + return ls.sessions[s], *addr + } + + // Load-guarded so the common case of a lane that will not come up is a + // plain read and cannot ping-pong the cache line these flags share. + if !ls.demand[s].Load() { + ls.demand[s].Store(true) + } + return nil, netip.AddrPort{} +} + +// laneTargetPortLocked returns the peer port lane s aims at. Only meaningful +// when peerPortCount is nonzero, which txLanes > 0 guarantees. +func (ls *laneSet) laneTargetPortLocked(s int) uint16 { + return ls.peerBasePort + uint16((s+int(ls.portOffset))%int(ls.peerPortCount)) +} + +// laneRetryDelay is the backoff after fails consecutive probe failures. +func laneRetryDelay(fails uint8) time.Duration { + d := laneRetryBase << min(fails, 4) + if d > laneRetryMax { + d = laneRetryMax + } + return d +} + +// noteAck records an acked probe for lane s, promoting the lane if it was down. +// gen must match the outstanding probe. Reports whether the lane was promoted. +func (ls *laneSet) noteAck(s int, gen uint8, now time.Time) bool { + if s <= 0 || s >= len(ls.probe) { + return false + } + + ls.mu.Lock() + defer ls.mu.Unlock() + + p := &ls.probe[s] + if p.sentAt.IsZero() || p.gen != gen { + // No probe outstanding, or an ack for a probe we have already given up + // on. Either way it says nothing about the lane's current path. + return false + } + + p.sentAt = time.Time{} + p.lastAck = now + p.fails = 0 + p.retryAt = time.Time{} + + if ls.txAddr[s].Load() != nil { + // Keepalive for a lane already up. + return false + } + + target := p.target + ls.txAddr[s].Store(&target) + return true +} + +// 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. +func (f *Interface) probeLanes(hostinfo *HostInfo, now time.Time, nb, out []byte) { + ls := hostinfo.lanes + if ls == nil || ls.txLanes < 2 { + return + } + + remote := hostinfo.GetRemote() + + ls.mu.Lock() + defer ls.mu.Unlock() + + if !remote.IsValid() { + // Relayed, or otherwise without a direct path. Lanes are direct-only, + // so drop them all; a later tick rebuilds if a direct path returns. + ls.resetLocked() + return + } + + if ls.peerAddr != remote.Addr() { + if ls.peerAddr.IsValid() { + // A roam is a new path, not a failure: forget the lanes built on + // the old one and let demand re-probe from a clean backoff. On the + // first pass there is nothing built yet, so just record the address. + ls.resetLocked() + } + ls.peerAddr = remote.Addr() + } + + for s := 1; s < ls.txLanes; s++ { + p := &ls.probe[s] + up := ls.txAddr[s].Load() != nil + + if !p.sentAt.IsZero() { + if now.Sub(p.sentAt) < laneProbeTimeout { + continue + } + + // An aged-out probe is a failure whether it was bringing the lane + // up or keeping it up. + p.sentAt = time.Time{} + p.fails = min(p.fails+1, laneMaxFails) + p.retryAt = now.Add(laneRetryDelay(p.fails)) + if up { + ls.txAddr[s].Store(nil) + hostinfo.logger(f.l).Info("Multiport lane demoted, probe unanswered", "lane", s) + } + continue + } + + if up { + if now.Sub(p.lastAck) < laneKeepalive { + continue + } + } else if now.Before(p.retryAt) || !ls.demand[s].Swap(false) { + continue + } + + p.gen++ + p.target = netip.AddrPortFrom(remote.Addr(), ls.laneTargetPortLocked(s)) + if f.sendLaneProbe(hostinfo, s, p.gen, p.target, nb, out) { + p.sentAt = now + } else { + p.fails = min(p.fails+1, laneMaxFails) + p.retryAt = now.Add(laneRetryDelay(p.fails)) + } + } +} + +// resetLocked takes every lane down and clears its probe state, without +// counting it as a failure. +// +// 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 +// exists again, while a lane whose routine has gone quiet stays down. +func (ls *laneSet) resetLocked() { + for s := 1; s < len(ls.probe); s++ { + ls.txAddr[s].Store(nil) + ls.probe[s] = laneProbeState{} + } +} + +// sendLaneProbe sends a probe on lane s to addr from writers[s]. 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 us, and the +// keys we derived for this lane match the ones it derived. Reports whether the +// probe made it onto the wire. +func (f *Interface) sendLaneProbe(hostinfo *HostInfo, s int, gen uint8, addr netip.AddrPort, nb, out []byte) bool { + ci := hostinfo.lanes.sessions[s] + if ci == nil || ci.eKey == nil { + return false + } + + if noiseutil.EncryptLockNeeded { + ci.writeLock.Lock() + } + c, ok := ci.NextMessageCounter() + if !ok { + if noiseutil.EncryptLockNeeded { + ci.writeLock.Unlock() + } + f.dropExhausted(hostinfo, c, "Dropping multiport lane probe, lane message counter is exhausted") + return false + } + + b := header.EncodeLane(out[:0], header.Version, header.Test, header.LaneProbe, hostinfo.remoteIndexId, c, uint8(s)) + b, err := ci.eKey.EncryptDanger(b, b, []byte{uint8(s), gen}, c, nb) + if noiseutil.EncryptLockNeeded { + ci.writeLock.Unlock() + } + if err != nil { + hostinfo.logger(f.l).Error("Failed to encrypt multiport lane probe", "error", err, "lane", s) + return false + } + + f.messageMetrics.Tx(header.Test, header.LaneProbe, 1) + if err := f.writers[s].WriteTo(b, addr); err != nil { + hostinfo.logger(f.l).Error("Failed to send multiport lane probe", "error", err, "lane", s, "udpAddr", addr) + return false + } + + if f.l.Enabled(context.Background(), slog.LevelDebug) { + hostinfo.logger(f.l).Debug("Multiport lane probe sent", "lane", s, "gen", gen, "udpAddr", addr) + } + return true +} + +// handleLaneProbe answers a peer's lane probe. The ack rides the base tunnel on +// purpose: a probe proves the peer's lane s works in its send direction, and +// answering on our own lane s would make the result depend on a second path +// that may be broken independently. +func (f *Interface) handleLaneProbe(hostinfo *HostInfo, lane uint8, payload []byte, rxc *rxContext) { + if lane == 0 || len(payload) < 2 { + return + } + + // Echo the header's lane rather than the payload's, so a peer cannot get us + // to vouch for a lane it did not actually probe. + f.send(header.Test, header.LaneProbeAck, hostinfo.ConnectionState, hostinfo, + []byte{lane, payload[1]}, rxc.nb, rxc.scratch[:0]) +} + +// handleLaneProbeAck promotes the lane a peer just acked. +func (f *Interface) handleLaneProbeAck(hostinfo *HostInfo, payload []byte) { + ls := hostinfo.lanes + if ls == nil || len(payload) < 2 { + return + } + + if ls.noteAck(int(payload[0]), payload[1], time.Now()) { + hostinfo.logger(f.l).Info("Multiport lane up", "lane", payload[0]) + } +} diff --git a/lanes_test.go b/lanes_test.go index 46e1a6f6..0f869ae9 100644 --- a/lanes_test.go +++ b/lanes_test.go @@ -8,6 +8,7 @@ import ( "github.com/gaissmai/bart" "github.com/slackhq/nebula/cert" "github.com/slackhq/nebula/config" + "github.com/slackhq/nebula/handshake" "github.com/slackhq/nebula/header" "github.com/slackhq/nebula/overlay/batch" "github.com/slackhq/nebula/overlay/overlaytest" @@ -18,33 +19,38 @@ import ( "github.com/stretchr/testify/require" ) -func newTestBaseHostInfo(vpnIp netip.Addr, localIdx, remoteIdx uint32, laneCount int) *HostInfo { - base := &HostInfo{ - vpnAddrs: []netip.Addr{vpnIp}, - localIndexId: localIdx, - remoteIndexId: remoteIdx, - remotes: NewRemoteList([]netip.Addr{vpnIp}, nil), - HandshakePacket: map[uint8][]byte{}, - } - base.SetRemote(netip.MustParseAddrPort("192.0.2.1:4242")) - base.lanes = newLaneState(laneCount, uint16(laneCount), 4242, 0) - return base +var ( + testMyAddr = netip.MustParseAddr("10.0.0.1") + testPeerAddr = netip.MustParseAddr("10.0.0.2") +) + +// newTestLaneSet derives a lane set from a real handshake result so the sessions +// hold usable keys. +func newTestLaneSet(t *testing.T, r *handshake.Result, myLanes int, peerPorts, peerBase, peerTxLanes uint32) *laneSet { + t.Helper() + r.PeerPortCount = peerPorts + r.PeerBasePort = peerBase + r.PeerTxLanes = peerTxLanes + ls, err := newLaneSet(r, myLanes, testMyAddr, testPeerAddr) + require.NoError(t, err) + return ls } -func newTestLaneHostInfo(base *HostInfo, laneIndex uint16, localIdx, remoteIdx uint32, owned bool) *HostInfo { - lane := &HostInfo{ - vpnAddrs: base.vpnAddrs, - localIndexId: localIdx, - remoteIndexId: remoteIdx, - remotes: NewRemoteList(base.vpnAddrs, nil), +func newTestLaneHostInfo(t *testing.T, r *handshake.Result, ls *laneSet) *HostInfo { + t.Helper() + cs, err := newConnectionStateFromResult(r) + require.NoError(t, err) + hi := &HostInfo{ + vpnAddrs: []netip.Addr{testPeerAddr}, + localIndexId: 100, + remoteIndexId: 200, + remotes: NewRemoteList([]netip.Addr{testPeerAddr}, nil), HandshakePacket: map[uint8][]byte{}, - sockIdx: int(laneIndex), - laneIndex: laneIndex, - laneOwned: owned, - parent: base, + ConnectionState: cs, + lanes: ls, } - lane.SetRemote(netip.MustParseAddrPort("192.0.2.1:4243")) - return lane + hi.SetRemote(netip.MustParseAddrPort("192.0.2.1:4242")) + return hi } func TestLanePortOffset(t *testing.T) { @@ -83,132 +89,149 @@ func TestLanePortOffset(t *testing.T) { } func TestLaneTargetPort(t *testing.T) { + ls := &laneSet{peerBasePort: 4242, peerPortCount: 4} + // No rotation: lane i targets base+i, wrapping past the peer's range. - ls := newLaneState(4, 4, 4242, 0) for i, want := range map[int]uint16{1: 4243, 2: 4244, 3: 4245, 5: 4243} { - assert.Equal(t, want, ls.laneTargetPort(i), "lane %d", i) + assert.Equal(t, want, ls.laneTargetPortLocked(i), "lane %d", i) } // Rotation shifts the whole mapping; the wrapped lane lands on the base // port itself, which is a valid distinct 4-tuple (our source port differs). - ls = newLaneState(4, 4, 4242, 3) + ls.portOffset = 3 for i, want := range map[int]uint16{1: 4242, 2: 4243, 3: 4244} { - assert.Equal(t, want, ls.laneTargetPort(i), "rotated lane %d", i) + assert.Equal(t, want, ls.laneTargetPortLocked(i), "rotated lane %d", i) } // Fewer peer ports than local lanes: rotation still spreads across all of // the peer's ports. - ls = newLaneState(16, 2, 4242, 1) - assert.Equal(t, uint16(4242), ls.laneTargetPort(1)) - assert.Equal(t, uint16(4243), ls.laneTargetPort(2)) + ls = &laneSet{peerBasePort: 4242, peerPortCount: 2, portOffset: 1} + assert.Equal(t, uint16(4242), ls.laneTargetPortLocked(1)) + assert.Equal(t, uint16(4243), ls.laneTargetPortLocked(2)) } -func TestLaneHostmapLifecycle(t *testing.T) { - l := test.NewLogger() - hostMap := newHostMap(l) - ifce := &Interface{l: l} // connectionManager nil is tolerated by unlockedAddLane +// A lane's keys are derived, not negotiated, so the whole design rests on the +// two sides landing on the same pair without exchanging anything. +func TestLaneKeyDerivationSymmetry(t *testing.T) { + initR, respR := runTestHandshake(t) - vpnIp := netip.MustParseAddr("172.1.1.2") - base := newTestBaseHostInfo(vpnIp, 100, 200, 4) + initLS := newTestLaneSet(t, initR, 4, 4, 4242, 4) + respLS := newTestLaneSet(t, respR, 4, 4, 4242, 4) + require.Len(t, initLS.sessions, 4) + assert.Nil(t, initLS.sessions[0], "lane 0 is the base session, not a derived one") - hostMap.Lock() - hostMap.unlockedAddHostInfo(base, ifce) - hostMap.Unlock() + nb := make([]byte, 12) + for s := 1; s < 4; s++ { + out := header.EncodeLane(make([]byte, 0, mtu), header.Version, header.Message, 0, 200, 1, uint8(s)) + ct, err := initLS.sessions[s].eKey.EncryptDanger(out, out, []byte("lane payload"), 1, nb) + require.NoError(t, err) - lane := newTestLaneHostInfo(base, 1, 101, 201, true) - hostMap.Lock() - hostMap.unlockedAddLane(lane, ifce) - hostMap.Unlock() - base.lanes.txLanes[1].Store(lane) + pt, err := respLS.sessions[s].Decrypt(test.NewLogger(), 1, ct, nb) + require.NoError(t, err, "lane %d keys did not match", s) + assert.Equal(t, []byte("lane payload"), pt) + } - // The lane is reachable by index (RX demux, recv_error) but never a Hosts primary. - assert.Equal(t, lane, hostMap.QueryIndex(101)) - assert.Equal(t, lane, hostMap.QueryReverseIndex(201)) - assert.Equal(t, base, hostMap.Hosts[vpnIp]) - - // A lane can never be promoted to primary. - hostMap.Lock() - assert.False(t, hostMap.unlockedMakePrimary(lane)) - hostMap.Unlock() - assert.Equal(t, base, hostMap.Hosts[vpnIp]) - - // Deleting the lane clears only its slot, applies backoff, and never - // reports "no more tunnels to peer" (final). - final := hostMap.DeleteHostInfo(lane) - assert.False(t, final) - assert.Nil(t, hostMap.QueryIndex(101)) - assert.Equal(t, base, hostMap.Hosts[vpnIp]) - assert.Nil(t, base.lanes.txLanes[1].Load()) - base.lanes.Lock() - assert.Equal(t, uint8(1), base.lanes.txFails[1]) - assert.False(t, base.lanes.txPending[1]) - assert.True(t, base.lanes.txRetryAt[1].After(time.Now())) - base.lanes.Unlock() - - // Idempotent: deleting again must not bump the backoff further. - hostMap.DeleteHostInfo(lane) - base.lanes.Lock() - assert.Equal(t, uint8(2), base.lanes.txFails[1]) // noteLaneFailure still runs, but slot CAS is a no-op - base.lanes.Unlock() + // Distinct lanes get distinct keys: lane 2's session must not open lane 1's + // ciphertext, or the header's lane index would be forgeable in effect. + out := header.EncodeLane(make([]byte, 0, mtu), header.Version, header.Message, 0, 200, 7, 1) + ct, err := initLS.sessions[1].eKey.EncryptDanger(out, out, []byte("lane payload"), 7, nb) + require.NoError(t, err) + _, err = respLS.sessions[2].Decrypt(test.NewLogger(), 7, ct, nb) + assert.Error(t, err) } -func TestLaneHostmapCascadeDelete(t *testing.T) { - l := test.NewLogger() - hostMap := newHostMap(l) - ifce := &Interface{l: l} +func TestNewLaneSetSizing(t *testing.T) { + initR, _ := runTestHandshake(t) - vpnIp := netip.MustParseAddr("172.1.1.3") - base := newTestBaseHostInfo(vpnIp, 300, 400, 4) + // A peer with no multiport advert gets no lanes at all. + ls, err := newLaneSet(&handshake.Result{}, 4, testMyAddr, testPeerAddr) + require.NoError(t, err) + assert.Nil(t, ls) - hostMap.Lock() - hostMap.unlockedAddHostInfo(base, ifce) - hostMap.Unlock() + // One lane means only the base tunnel, which is not a lane set. + ls, err = newLaneSet(&handshake.Result{PeerPortCount: 4, PeerTxLanes: 1}, 1, testMyAddr, testPeerAddr) + require.NoError(t, err) + assert.Nil(t, ls) - owned := newTestLaneHostInfo(base, 1, 301, 401, true) - peer := newTestLaneHostInfo(base, 2, 302, 402, false) - hostMap.Lock() - hostMap.unlockedAddLane(owned, ifce) - hostMap.unlockedAddLane(peer, ifce) - hostMap.Unlock() - base.lanes.txLanes[1].Store(owned) - base.lanes.Lock() - base.lanes.peerLanes = append(base.lanes.peerLanes, peer) - base.lanes.Unlock() + // Sessions cover both directions: enough for everything the peer may send, + // even though we may only send on a few. + ls = newTestLaneSet(t, initR, 2, 8, 4242, 6) + assert.Len(t, ls.sessions, 6, "sessions must cover the peer's tx lanes") + assert.Equal(t, 2, ls.txLanes, "we may only send on our own lanes") - // Deleting the base takes the whole lane family with it. - final := hostMap.DeleteHostInfo(base) - assert.True(t, final) - assert.Nil(t, hostMap.QueryIndex(300)) - assert.Nil(t, hostMap.QueryIndex(301)) - assert.Nil(t, hostMap.QueryIndex(302)) - assert.Nil(t, hostMap.Hosts[vpnIp]) + // Our tx lanes are clamped to the ports the peer actually bound: a lane + // aimed past the peer's range would land on some unrelated socket. + ls = newTestLaneSet(t, initR, 8, 3, 4242, 8) + assert.Equal(t, 3, ls.txLanes) + assert.Len(t, ls.sessions, 8) } -// Regression: deleting a hostinfo whose pending entry is NOT the one recorded -// in vpnIps (e.g. a lane, whose vpnAddrs alias the base's) must not evict a -// concurrently pending base handshake for the same address. -func TestHandshakeManagerVpnIpsIdentityDelete(t *testing.T) { - l := test.NewLogger() - hostMap := newHostMap(l) - lh := newTestLighthouse() - hm := NewHandshakeManager(l, hostMap, lh, &udp.NoopConn{}, defaultHandshakeConfig) +func TestLaneTxGate(t *testing.T) { + initR, _ := runTestHandshake(t) + ls := newTestLaneSet(t, initR, 4, 4, 4242, 4) - vpnIp := netip.MustParseAddr("172.1.1.4") - pendingBase := hm.StartHandshake(vpnIp, nil) - require.NotNil(t, pendingBase) + // A down lane hands back nothing and raises demand, which is what gets it + // probed. + ci, addr := ls.txLane(1) + assert.Nil(t, ci) + assert.False(t, addr.IsValid()) + assert.True(t, ls.demand[1].Load()) + assert.False(t, ls.demand[2].Load(), "demand raised on an untouched lane") - other := &HostInfo{vpnAddrs: []netip.Addr{vpnIp}, localIndexId: 999} - hm.DeleteHostInfo(other) + // Promotion publishes the session and destination together. + want := netip.MustParseAddrPort("192.0.2.1:4243") + ls.txAddr[1].Store(&want) + ls.demand[1].Store(false) + ci, addr = ls.txLane(1) + assert.Equal(t, ls.sessions[1], ci) + assert.Equal(t, want, addr) + assert.False(t, ls.demand[1].Load(), "a hit must not raise demand") - // The pending base handshake must still be tracked. - assert.Equal(t, pendingBase, hm.QueryVpnAddr(vpnIp)) - - // And deleting the actual owner still works. - hm.DeleteHostInfo(pendingBase) - assert.Nil(t, hm.QueryVpnAddr(vpnIp)) + // Lane 0 is the base tunnel and lanes at or above txLanes are receive-only. + ci, _ = ls.txLane(0) + assert.Nil(t, ci) + ci, _ = ls.txLane(4) + assert.Nil(t, ci) } -func newLaneTestConnectionManager(hostMap *HostMap) (*connectionManager, *Interface) { +func TestLaneSessionLookup(t *testing.T) { + initR, _ := runTestHandshake(t) + ls := newTestLaneSet(t, initR, 3, 4, 4242, 3) + hi := newTestLaneHostInfo(t, initR, ls) + + assert.Nil(t, hi.laneSession(0), "lane 0 is the base session") + assert.Equal(t, ls.sessions[2], hi.laneSession(2)) + assert.Nil(t, hi.laneSession(3), "a lane we never derived") + assert.Nil(t, hi.laneSession(255)) + + // A peer without lanes answers nil for every lane rather than panicking. + bare := &HostInfo{} + assert.Nil(t, bare.laneSession(1)) +} + +func TestMaxMessageCounter(t *testing.T) { + initR, _ := runTestHandshake(t) + ls := newTestLaneSet(t, initR, 3, 4, 4242, 3) + hi := newTestLaneHostInfo(t, initR, ls) + + hi.ConnectionState.messageCounter.Store(5) + assert.Equal(t, uint64(5), hi.maxMessageCounter()) + + // A lane past the base is what the rehandshake threshold has to notice: the + // base counter would sit still while the lane burns through its nonces. + ls.sessions[2].messageCounter.Store(9000) + assert.Equal(t, uint64(9000), hi.maxMessageCounter()) + + assert.Equal(t, uint64(0), (&HostInfo{}).maxMessageCounter()) +} + +func TestLaneRetryDelay(t *testing.T) { + assert.Equal(t, laneRetryBase, laneRetryDelay(0)) + assert.Equal(t, 2*laneRetryBase, laneRetryDelay(1)) + assert.Equal(t, laneRetryMax, laneRetryDelay(laneMaxFails), "backoff must saturate") +} + +func newLaneTestInterface(hostMap *HostMap) *Interface { l := test.NewLogger() lh := newTestLighthouse() cs := &CertState{ @@ -226,86 +249,225 @@ func newLaneTestConnectionManager(hostMap *HostMap) (*connectionManager, *Interf pki: &PKI{}, handshakeManager: NewHandshakeManager(l, hostMap, lh, &udp.NoopConn{}, defaultHandshakeConfig), myVpnNetworksTable: new(bart.Lite), + messageMetrics: newMessageMetricsOnlyRecvError(), + writers: []udp.Conn{&udp.NoopConn{}, &udp.NoopConn{}, &udp.NoopConn{}, &udp.NoopConn{}}, l: l, } ifce.pki.cs.Store(cs) - conf := config.NewC(test.NewLogger()) - punchy := NewPunchyFromConfig(test.NewLogger(), conf, nil) - cm := newConnectionManagerFromConfig(test.NewLogger(), conf, hostMap, punchy) + conf := config.NewC(l) + punchy := NewPunchyFromConfig(l, conf, nil) + cm := newConnectionManagerFromConfig(l, conf, hostMap, punchy) cm.intf = ifce ifce.connectionManager = cm ifce.handshakeManager.f = ifce - return cm, ifce + return ifce } -func TestLaneTrafficDecision(t *testing.T) { +// The full TX lifecycle of a lane: demand -> probe -> ack -> up, then an +// unanswered keepalive -> demoted. +func TestLaneProbeLifecycle(t *testing.T) { hostMap := newHostMap(test.NewLogger()) - cm, ifce := newLaneTestConnectionManager(hostMap) + ifce := newLaneTestInterface(hostMap) - vpnIp := netip.MustParseAddr("172.1.1.5") - base := newTestBaseHostInfo(vpnIp, 500, 600, 4) - base.ConnectionState = &ConnectionState{} - hostMap.Lock() - hostMap.unlockedAddHostInfo(base, ifce) - hostMap.Unlock() - - lane := newTestLaneHostInfo(base, 1, 501, 601, true) - lane.ConnectionState = &ConnectionState{} - hostMap.Lock() - hostMap.unlockedAddLane(lane, ifce) - hostMap.Unlock() - base.lanes.txLanes[1].Store(lane) + initR, _ := runTestHandshake(t) + ls := newTestLaneSet(t, initR, 4, 4, 5353, 4) + hi := newTestLaneHostInfo(t, initR, ls) + nb := make([]byte, 12) + out := make([]byte, mtu) now := time.Now() - // A lane with inbound traffic is alive and never swaps primary or - // migrates relays. - lane.markIn() - decision, resolved, _ := cm.makeTrafficDecision(lane.localIndexId, now) - assert.Equal(t, doNothing, decision) - assert.Equal(t, lane, resolved) - assert.False(t, lane.isPendingDeletion()) + // No demand: nothing is probed, so a peer we barely talk to costs nothing + // beyond its base tunnel. + ifce.probeLanes(hi, now, nb, out) + ls.mu.Lock() + for s := 1; s < 4; s++ { + assert.True(t, ls.probe[s].sentAt.IsZero(), "lane %d probed without demand", s) + } + ls.mu.Unlock() - // An idle lane gets an active keepalive test... - decision, _, _ = cm.makeTrafficDecision(lane.localIndexId, now) - assert.Equal(t, sendTestPacket, decision) - assert.True(t, lane.isPendingDeletion()) + // Demand on lane 1 alone probes lane 1 alone, aimed at the peer's port for + // that lane. + ls.demand[1].Store(true) + ifce.probeLanes(hi, now, nb, out) + ls.mu.Lock() + require.False(t, ls.probe[1].sentAt.IsZero(), "demand did not produce a probe") + assert.True(t, ls.probe[2].sentAt.IsZero()) + wantPort := ls.laneTargetPortLocked(1) + assert.Equal(t, netip.AddrPortFrom(netip.MustParseAddr("192.0.2.1"), wantPort), ls.probe[1].target) + gen := ls.probe[1].gen + assert.False(t, ls.demand[1].Load(), "the probe did not consume the demand") + ls.mu.Unlock() - // ...and is declared dead when the test goes unanswered. - decision, _, _ = cm.makeTrafficDecision(lane.localIndexId, now) - assert.Equal(t, deleteTunnel, decision) + // The lane stays down until the ack lands, and a stale generation cannot + // bring it up. + assert.Nil(t, ls.txAddr[1].Load()) + assert.False(t, ls.noteAck(1, gen+1, now), "an ack for a superseded probe promoted the lane") + assert.Nil(t, ls.txAddr[1].Load()) + + // The matching ack promotes it, and the destination is the probed target. + assert.True(t, ls.noteAck(1, gen, now)) + addr := ls.txAddr[1].Load() + require.NotNil(t, addr) + assert.Equal(t, netip.AddrPortFrom(netip.MustParseAddr("192.0.2.1"), wantPort), *addr) + + // A second ack is a keepalive, not a promotion. + ls.mu.Lock() + ls.probe[1].sentAt = now + ls.mu.Unlock() + assert.False(t, ls.noteAck(1, gen, now)) + assert.NotNil(t, ls.txAddr[1].Load()) + + // An up lane is left alone until the keepalive comes due. + ifce.probeLanes(hi, now, nb, out) + ls.mu.Lock() + assert.True(t, ls.probe[1].sentAt.IsZero(), "an up lane was re-probed early") + ls.mu.Unlock() + + // Past the keepalive it re-proves its path... + now = now.Add(laneKeepalive + time.Second) + ifce.probeLanes(hi, now, nb, out) + ls.mu.Lock() + require.False(t, ls.probe[1].sentAt.IsZero(), "keepalive did not probe") + ls.mu.Unlock() + assert.NotNil(t, ls.txAddr[1].Load(), "lane demoted before its probe aged out") + + // ...and an unanswered keepalive demotes it with backoff, so the routine + // falls back to the base tunnel. + now = now.Add(laneProbeTimeout + time.Second) + ifce.probeLanes(hi, now, nb, out) + assert.Nil(t, ls.txAddr[1].Load(), "unanswered keepalive did not demote the lane") + ls.mu.Lock() + assert.Equal(t, uint8(1), ls.probe[1].fails) + assert.True(t, ls.probe[1].retryAt.After(now)) + ls.mu.Unlock() + + // The backoff holds even with fresh demand. + ls.demand[1].Store(true) + ifce.probeLanes(hi, now, nb, out) + ls.mu.Lock() + assert.True(t, ls.probe[1].sentAt.IsZero(), "backoff was ignored") + ls.mu.Unlock() } -func TestBaseInactiveConsidersLanes(t *testing.T) { +// A roam is a new path with no derivable relationship to the old lane ports, so +// every lane has to be rebuilt rather than moved. +func TestLaneProbeRoamResets(t *testing.T) { hostMap := newHostMap(test.NewLogger()) - cm, _ := newLaneTestConnectionManager(hostMap) - cm.dropInactive.Store(true) - cm.inactivityTimeout.Store(int64(10 * time.Minute)) + ifce := newLaneTestInterface(hostMap) + + initR, _ := runTestHandshake(t) + ls := newTestLaneSet(t, initR, 4, 4, 5353, 4) + hi := newTestLaneHostInfo(t, initR, ls) now := time.Now() - vpnIp := netip.MustParseAddr("172.1.1.6") - base := newTestBaseHostInfo(vpnIp, 700, 800, 4) - base.lastUsed = now.Add(-time.Hour) + nb := make([]byte, 12) + out := make([]byte, mtu) - // Base alone: inactive. - _, inactive := cm.isInactive(base, now) - assert.True(t, inactive) + ls.demand[1].Store(true) + ifce.probeLanes(hi, now, nb, out) + ls.mu.Lock() + gen := ls.probe[1].gen + ls.mu.Unlock() + require.True(t, ls.noteAck(1, gen, now)) + require.NotNil(t, ls.txAddr[1].Load()) - // A recently used lane keeps the base alive. - lane := newTestLaneHostInfo(base, 1, 701, 801, true) - lane.lastUsed = now.Add(-time.Minute) - base.lanes.txLanes[1].Store(lane) - _, inactive = cm.isInactive(base, now) - assert.False(t, inactive) + // New remote address: the lane is taken down, not retargeted. + hi.SetRemote(netip.MustParseAddrPort("198.51.100.7:4242")) + ifce.probeLanes(hi, now, nb, out) + assert.Nil(t, ls.txAddr[1].Load(), "lane survived a roam") + ls.mu.Lock() + assert.Zero(t, ls.probe[1].fails, "a roam is not a lane failure") + assert.Equal(t, netip.MustParseAddr("198.51.100.7"), ls.peerAddr) + ls.mu.Unlock() - // Peer-owned lanes count too. - base.lanes.txLanes[1].Store(nil) - base.lanes.Lock() - base.lanes.peerLanes = append(base.lanes.peerLanes, lane) - base.lanes.Unlock() - _, inactive = cm.isInactive(base, now) - assert.False(t, inactive) + // Relayed (no direct remote) means no lanes at all. + ls.demand[1].Store(true) + ifce.probeLanes(hi, now, nb, out) + ls.mu.Lock() + gen = ls.probe[1].gen + ls.mu.Unlock() + require.True(t, ls.noteAck(1, gen, now)) + hi.SetRemote(netip.AddrPort{}) + ifce.probeLanes(hi, now, nb, out) + assert.Nil(t, ls.txAddr[1].Load(), "lane survived losing the direct path") +} + +// 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) { + hostMap := newHostMap(test.NewLogger()) + ifce := newLaneTestInterface(hostMap) + + initR, respR := runTestHandshake(t) + ls := newTestLaneSet(t, respR, 4, 4, 4242, 4) + hi := newTestLaneHostInfo(t, respR, ls) + + rxc := &rxContext{nb: make([]byte, 12), scratch: make([]byte, mtu)} + sent := &recordingUdpConn{} + ifce.writers = []udp.Conn{sent, &udp.NoopConn{}, &udp.NoopConn{}, &udp.NoopConn{}} + + // The payload's lane is ignored in favour of the header's. + ifce.handleLaneProbe(hi, 2, []byte{3, 42}, rxc) + require.Len(t, sent.bufs, 1, "the ack must ride the base tunnel's socket") + + h := &header.H{} + require.NoError(t, h.Parse(sent.bufs[0])) + 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") + + pt, err := newTestLaneSet(t, initR, 4, 4, 4242, 4).sessions[1].dKey.DecryptDanger( + nil, sent.bufs[0][:header.Len], sent.bufs[0][header.Len:], h.MessageCounter, make([]byte, 12)) + _ = pt + assert.Error(t, err, "the ack must not be readable with a lane key") + + // A probe claiming lane 0 or with a truncated payload is answered with + // nothing at all. + sent.bufs = nil + ifce.handleLaneProbe(hi, 0, []byte{1, 2}, rxc) + ifce.handleLaneProbe(hi, 1, []byte{1}, rxc) + assert.Empty(t, sent.bufs) +} + +func TestHandleLaneProbeAck(t *testing.T) { + hostMap := newHostMap(test.NewLogger()) + ifce := newLaneTestInterface(hostMap) + + initR, _ := runTestHandshake(t) + ls := newTestLaneSet(t, initR, 4, 4, 5353, 4) + hi := newTestLaneHostInfo(t, initR, ls) + + now := time.Now() + ls.demand[1].Store(true) + ifce.probeLanes(hi, now, make([]byte, 12), make([]byte, mtu)) + ls.mu.Lock() + gen := ls.probe[1].gen + ls.mu.Unlock() + + // A short or out-of-range ack is ignored, and a peer without lanes does not + // panic the handler. + ifce.handleLaneProbeAck(hi, []byte{1}) + ifce.handleLaneProbeAck(hi, []byte{99, gen}) + ifce.handleLaneProbeAck(&HostInfo{}, []byte{1, gen}) + assert.Nil(t, ls.txAddr[1].Load()) + + ifce.handleLaneProbeAck(hi, []byte{1, gen}) + assert.NotNil(t, ls.txAddr[1].Load()) +} + +// recordingUdpConn records what was written to it, one datagram per write. +type recordingUdpConn struct { + udp.NoopConn + bufs [][]byte + dsts []netip.AddrPort +} + +func (c *recordingUdpConn) WriteTo(b []byte, addr netip.AddrPort) error { + c.bufs = append(c.bufs, append([]byte(nil), b...)) + c.dsts = append(c.dsts, addr) + return nil } // recordingBatchWriter satisfies batch's writer interface and records what @@ -325,21 +487,12 @@ func (w *recordingBatchWriter) WriteBatch(bufs [][]byte, addrs []netip.AddrPort) func TestSendInsideMessageLaneSwap(t *testing.T) { hostMap := newHostMap(test.NewLogger()) - cm, ifce := newLaneTestConnectionManager(hostMap) - _ = cm + ifce := newLaneTestInterface(hostMap) - vpnIp := netip.MustParseAddr("172.1.1.7") - base := newTestBaseHostInfo(vpnIp, 900, 1000, 4) - lane := newTestLaneHostInfo(base, 1, 901, 1001, true) - - // Real cipher states from a real handshake so encryption works. - baseInit, _ := runTestHandshake(t) - laneInit, _ := runTestHandshake(t) - var err error - base.ConnectionState, err = newConnectionStateFromResult(baseInit) - require.NoError(t, err) - lane.ConnectionState, err = newConnectionStateFromResult(laneInit) - require.NoError(t, err) + initR, respR := runTestHandshake(t) + ls := newTestLaneSet(t, initR, 4, 4, 5353, 4) + hi := newTestLaneHostInfo(t, initR, ls) + peerLS := newTestLaneSet(t, respR, 4, 4, 4242, 4) baseWriter := &recordingBatchWriter{} laneWriter := &recordingBatchWriter{} @@ -356,41 +509,60 @@ func TestSendInsideMessageLaneSwap(t *testing.T) { pkt := tio.Packet{Bytes: []byte{0x45, 0, 0, 4, 1, 2, 3, 4}} nb := make([]byte, 12) - // With the lane published, slot-1 traffic uses the lane session and the - // lane batch. - base.lanes.txLanes[1].Store(lane) - ifce.sendInsideMessage(base, pkt, nb, tx1) + // A down lane rides the base tunnel and asks for a probe. + ifce.sendInsideMessage(hi, pkt, nb, tx1) tx1.flush(ifce) - require.Len(t, laneWriter.bufs, 1) - require.Empty(t, baseWriter.bufs) - assert.Equal(t, lane.GetRemote(), laneWriter.dsts[0]) + require.Len(t, baseWriter.bufs, 1) + assert.Empty(t, laneWriter.bufs) + assert.True(t, ls.demand[1].Load(), "a miss did not raise demand") + assert.False(t, ls.demand[2].Load(), "demand raised on an untouched lane") h := &header.H{} - require.NoError(t, h.Parse(laneWriter.bufs[0])) - assert.Equal(t, lane.remoteIndexId, h.RemoteIndex) + require.NoError(t, h.Parse(baseWriter.bufs[0])) + assert.Equal(t, uint8(0), h.Lane()) - // An overflow routine sharing slot 1 (multiport.lanes < routines) rides - // the same lane session. + // Once the lane is up, slot-1 traffic rides the lane session, the lane + // socket and the lane's destination, tagged with the lane index. + laneRemote := netip.MustParseAddrPort("192.0.2.1:5354") + ls.txAddr[1].Store(&laneRemote) + ls.demand[1].Store(false) + ifce.sendInsideMessage(hi, pkt, nb, tx1) + tx1.flush(ifce) + require.Len(t, laneWriter.bufs, 1) + assert.Equal(t, laneRemote, laneWriter.dsts[0]) + assert.False(t, ls.demand[1].Load(), "a hit raised demand") + + require.NoError(t, h.Parse(laneWriter.bufs[0])) + assert.Equal(t, uint8(1), h.Lane()) + assert.Equal(t, hi.remoteIndexId, h.RemoteIndex) + + // And the peer's derived lane-1 session is what opens it. + pt, err := peerLS.sessions[1].Decrypt(test.NewLogger(), h.MessageCounter, laneWriter.bufs[0], nb) + require.NoError(t, err) + assert.Equal(t, pkt.Bytes, pt) + + // An overflow routine sharing slot 1 (multiport.lanes < routines) rides the + // same lane session. tx1b := newTx(1) - ifce.sendInsideMessage(base, pkt, nb, tx1b) + ifce.sendInsideMessage(hi, pkt, nb, tx1b) tx1b.flush(ifce) require.Len(t, laneWriter.bufs, 2) require.NoError(t, h.Parse(laneWriter.bufs[1])) - assert.Equal(t, lane.remoteIndexId, h.RemoteIndex) + assert.Equal(t, uint8(1), h.Lane()) - // Slot 2 has no lane: base tunnel, base batch. - ifce.sendInsideMessage(base, pkt, nb, tx2) + // Slot 2's lane is still down: base tunnel, base batch, lane 0. + ifce.sendInsideMessage(hi, pkt, nb, tx2) tx2.flush(ifce) - require.Len(t, baseWriter.bufs, 1) - assert.Equal(t, base.GetRemote(), baseWriter.dsts[0]) - require.NoError(t, h.Parse(baseWriter.bufs[0])) - assert.Equal(t, base.remoteIndexId, h.RemoteIndex) - - // Lane death: slot cleared, instant fallback to base. - base.lanes.txLanes[1].Store(nil) - ifce.sendInsideMessage(base, pkt, nb, tx1) - tx1.flush(ifce) require.Len(t, baseWriter.bufs, 2) + assert.Equal(t, hi.GetRemote(), baseWriter.dsts[1]) + require.NoError(t, h.Parse(baseWriter.bufs[1])) + assert.Equal(t, uint8(0), h.Lane()) + + // Demotion falls back to the base tunnel on the very next packet. + ls.txAddr[1].Store(nil) + ifce.sendInsideMessage(hi, pkt, nb, tx1) + tx1.flush(ifce) + require.Len(t, baseWriter.bufs, 3) require.Len(t, laneWriter.bufs, 2) } @@ -414,174 +586,25 @@ func TestLaneSlotFor(t *testing.T) { } } -func TestCompleteLaneResponder(t *testing.T) { - hostMap := newHostMap(test.NewLogger()) - _, ifce := newLaneTestConnectionManager(hostMap) - ifce.writers = []udp.Conn{&udp.NoopConn{}, &udp.NoopConn{}, &udp.NoopConn{}, &udp.NoopConn{}} - ifce.messageMetrics = newMessageMetricsOnlyRecvError() +// Regression: deleting a hostinfo whose pending entry is NOT the one recorded +// in vpnIps must not evict a concurrently pending handshake for that address. +func TestHandshakeManagerVpnIpsIdentityDelete(t *testing.T) { + l := test.NewLogger() + hostMap := newHostMap(l) + lh := newTestLighthouse() + hm := NewHandshakeManager(l, hostMap, lh, &udp.NoopConn{}, defaultHandshakeConfig) - hm := ifce.handshakeManager - hm.config.laneCount = 4 - hm.config.lanePortCount = 4 - hm.config.laneBasePort = 4242 + vpnIp := netip.MustParseAddr("172.1.1.4") + pendingBase := hm.StartHandshake(vpnIp, nil) + require.NotNil(t, pendingBase) - // A real handshake supplies usable keys and a peer cert. - _, respR := runTestHandshake(t) - respR.PeerLaneIndex = 2 - respR.PeerPortCount = 4 - respR.PeerBasePort = 5353 + other := &HostInfo{vpnAddrs: []netip.Addr{vpnIp}, localIndexId: 999} + hm.DeleteHostInfo(other) - via := ViaSender{UdpAddr: netip.MustParseAddrPort("192.0.2.9:5355"), SockIdx: 2} - packet := make([]byte, header.Len+8) - copy(packet[header.Len:], []byte("stage0!!")) - vpnAddrs := []netip.Addr{netip.MustParseAddr("172.1.1.9")} + // The pending base handshake must still be tracked. + assert.Equal(t, pendingBase, hm.QueryVpnAddr(vpnIp)) - // No base tunnel: the lane handshake is dropped, nothing registered. - hm.completeLaneResponder(via, packet, []byte("resp"), respR, vpnAddrs) - assert.Nil(t, hostMap.QueryIndex(respR.LocalIndex)) - - // With a live base the lane attaches to it. - base := newTestBaseHostInfo(vpnAddrs[0], 1300, 1400, 4) - base.ConnectionState = &ConnectionState{} - hostMap.Lock() - hostMap.unlockedAddHostInfo(base, ifce) - hostMap.Unlock() - - hm.completeLaneResponder(via, packet, []byte("resp"), respR, vpnAddrs) - lane := hostMap.QueryIndex(respR.LocalIndex) - require.NotNil(t, lane) - assert.True(t, lane.isLane()) - assert.False(t, lane.laneOwned) - assert.Equal(t, uint16(2), lane.laneIndex) - assert.Equal(t, 2, lane.sockIdx) - assert.Equal(t, via.UdpAddr, lane.GetRemote()) - assert.Equal(t, base, hostMap.Hosts[vpnAddrs[0]], "lane must not displace the base as primary") - base.lanes.Lock() - assert.Len(t, base.lanes.peerLanes, 1) - base.lanes.Unlock() - - // A byte-identical stage-0 retransmit resends the cached response and - // must not register a second lane. - hm.completeLaneResponder(via, packet, []byte("resp"), respR, vpnAddrs) - base.lanes.Lock() - assert.Len(t, base.lanes.peerLanes, 1) - base.lanes.Unlock() - - // An out-of-range lane index is refused. - respR2 := *respR - respR2.PeerLaneIndex = 9 - respR2.LocalIndex = respR.LocalIndex + 1 - hm.completeLaneResponder(via, packet, []byte("resp"), &respR2, vpnAddrs) - assert.Nil(t, hostMap.QueryIndex(respR2.LocalIndex)) -} - -func TestEnsureLanesBackoffOnStage0Failure(t *testing.T) { - hostMap := newHostMap(test.NewLogger()) - _, ifce := newLaneTestConnectionManager(hostMap) - - // laneCount enables multiport in the manager; the dummy CertState has no - // credential, so stage-0 construction must fail and release the slot with - // backoff rather than leaving it claimed forever. - hm := ifce.handshakeManager - hm.config.laneCount = 4 - hm.config.lanePortCount = 4 - hm.config.laneBasePort = 4242 - - vpnIp := netip.MustParseAddr("172.1.1.8") - base := newTestBaseHostInfo(vpnIp, 1100, 1200, 4) - - for i := 1; i < 4; i++ { - base.lanes.noteLaneDemand(i) - } - hm.EnsureLanes(base) - - base.lanes.Lock() - defer base.lanes.Unlock() - for i := 1; i < 4; i++ { - assert.False(t, base.lanes.txPending[i], "slot %d still pending", i) - assert.Equal(t, uint8(1), base.lanes.txFails[i], "slot %d fails", i) - assert.True(t, base.lanes.txRetryAt[i].After(time.Now()), "slot %d retryAt", i) - } -} - -// Lanes are demand-driven: a base tunnel with no data-plane interest in a slot -// must not start a handshake for it, and one demand must not turn into an -// unbounded retry loop. -func TestEnsureLanesLazy(t *testing.T) { - hostMap := newHostMap(test.NewLogger()) - _, ifce := newLaneTestConnectionManager(hostMap) - - hm := ifce.handshakeManager - hm.config.laneCount = 4 - hm.config.lanePortCount = 4 - hm.config.laneBasePort = 4242 - - base := newTestBaseHostInfo(netip.MustParseAddr("172.1.1.9"), 1300, 1400, 4) - ls := base.lanes - - // No demand: nothing is attempted, so nothing fails or backs off either. - hm.EnsureLanes(base) - ls.Lock() - for i := 1; i < 4; i++ { - assert.False(t, ls.txPending[i], "slot %d claimed without demand", i) - assert.Zero(t, ls.txFails[i], "slot %d attempted without demand", i) - assert.True(t, ls.txRetryAt[i].IsZero(), "slot %d backed off without demand", i) - } - ls.Unlock() - - // Demand on one slot starts that slot alone. (Stage 0 fails on the test - // CertState, so the observable effect is a failure charged to slot 2.) - ls.noteLaneDemand(2) - hm.EnsureLanes(base) - ls.Lock() - assert.Equal(t, uint8(1), ls.txFails[2]) - assert.Zero(t, ls.txFails[1], "untouched slot attempted") - assert.Zero(t, ls.txFails[3], "untouched slot attempted") - // The attempt consumed the demand, so a later tick past the backoff does - // not retry a lane nobody is asking for any more. - ls.txRetryAt[2] = time.Time{} - ls.Unlock() - hm.EnsureLanes(base) - ls.Lock() - assert.Equal(t, uint8(1), ls.txFails[2], "consumed demand was retried") - ls.Unlock() - - // Slot 0 is the base tunnel and is never a lane. - ls.noteLaneDemand(0) - assert.False(t, ls.takeLaneDemand(0)) -} - -// A TX miss on an empty slot is what asks for the lane; a hit must not, and a -// relay-only peer never gets that far. -func TestSendInsideMessageRecordsLaneDemand(t *testing.T) { - hostMap := newHostMap(test.NewLogger()) - _, ifce := newLaneTestConnectionManager(hostMap) - - base := newTestBaseHostInfo(netip.MustParseAddr("172.1.1.10"), 1500, 1600, 4) - lane := newTestLaneHostInfo(base, 1, 1501, 1601, true) - - init, _ := runTestHandshake(t) - cs, err := newConnectionStateFromResult(init) - require.NoError(t, err) - base.ConnectionState = cs - lane.ConnectionState = cs - - writer := &recordingBatchWriter{} - newTx := func(laneSlot int) *txQueue { - sb := batch.NewSendBatch(writer, batch.SendBatchCap, 1<<16) - return &txQueue{laneSlot: laneSlot, base: sb, lane: sb} - } - pkt := tio.Packet{Bytes: []byte{0x45, 0, 0, 4, 1, 2, 3, 4}} - nb := make([]byte, 12) - - // Empty slot: the send rides the base tunnel and flags demand for slot 1. - ifce.sendInsideMessage(base, pkt, nb, newTx(1)) - assert.True(t, base.lanes.txDemand[1].Load(), "miss did not raise demand") - assert.False(t, base.lanes.txDemand[2].Load(), "demand raised on an unused slot") - - // Established lane: the send uses it and asks for nothing. - base.lanes.txDemand[1].Store(false) - base.lanes.txLanes[1].Store(lane) - ifce.sendInsideMessage(base, pkt, nb, newTx(1)) - assert.False(t, base.lanes.txDemand[1].Load(), "hit raised demand") + // And deleting the actual owner still works. + hm.DeleteHostInfo(pendingBase) + assert.Nil(t, hm.QueryVpnAddr(vpnIp)) } diff --git a/main.go b/main.go index c70372ba..5ba52191 100644 --- a/main.go +++ b/main.go @@ -15,6 +15,7 @@ import ( "github.com/slackhq/nebula/config" "github.com/slackhq/nebula/cpupick" + "github.com/slackhq/nebula/header" "github.com/slackhq/nebula/noiseutil" "github.com/slackhq/nebula/overlay" "github.com/slackhq/nebula/sshd" @@ -315,6 +316,12 @@ func Main(c *config.C, configTest bool, buildVersion string, l *slog.Logger, dev if lanes <= 0 || lanes > routines { lanes = routines } + if lanes > header.MaxLane+1 { + // The lane index rides in one byte of the nebula header, so lanes + // above that are unaddressable. + l.Warn("multiport.lanes clamped to header limit", "lanes", lanes, "limit", header.MaxLane+1) + lanes = header.MaxLane + 1 + } handshakeConfig.laneCount = lanes handshakeConfig.lanePortCount = uint16(routines) handshakeConfig.laneBasePort = uint16(port) diff --git a/noiseutil/cipher_state.go b/noiseutil/cipher_state.go index 5f10d004..2430cb9a 100644 --- a/noiseutil/cipher_state.go +++ b/noiseutil/cipher_state.go @@ -1,6 +1,7 @@ package noiseutil import ( + "crypto/cipher" "errors" "fmt" "math" @@ -52,3 +53,29 @@ func NewCipherState(s *noise.CipherState, cipherFunc noise.CipherFunc) CipherSta panic(fmt.Sprintf("noiseutil: unsupported cipher %q", cipherFunc.CipherName())) } } + +// NewCipherStateFromKey builds a data-plane CipherState directly from raw key +// bytes, bypassing the noise handshake. Multiport lanes use this to key a +// derived session off the base tunnel's negotiated keys; the caller owns the +// guarantee that key is used with exactly one CipherState so nonces never +// repeat. +func NewCipherStateFromKey(key [32]byte, cipherFunc noise.CipherFunc) CipherState { + c := cipherFunc.Cipher(key) + if cs, ok := c.(CipherState); ok { + return cs + } + + aead, ok := c.(cipher.AEAD) + if !ok { + panic(fmt.Sprintf("noiseutil: cipher %q does not expose an AEAD", cipherFunc.CipherName())) + } + + switch cipherFunc.CipherName() { + case noise.CipherAESGCM.CipherName(): + return &CipherStateAESGCM{c: aead} + case noise.CipherChaChaPoly.CipherName(): + return &CipherStateChaChaPoly{c: aead} + default: + panic(fmt.Sprintf("noiseutil: unsupported cipher %q", cipherFunc.CipherName())) + } +} diff --git a/outside.go b/outside.go index 299c5fd2..1a0e8053 100644 --- a/outside.go +++ b/outside.go @@ -107,7 +107,35 @@ func (f *Interface) readOutsidePackets(via ViaSender, packet []byte, rxc *rxCont return } - if len(packet) < header.Len+hostinfo.ConnectionState.dKey.Overhead() { + // Which session decrypts this packet is the lane index in the header. Lane 0 + // is the base tunnel; a higher lane is one of the sessions derived from it. + ci := hostinfo.ConnectionState + lane := h.Lane() + if lane != 0 { + ci = hostinfo.laneSession(lane) + if ci == nil { + // A lane we have no session for: a stale lane from a tunnel that has + // since rolled, or a peer sending above what it advertised. Dropping + // silently is right for both — a recv_error would tear down a + // perfectly good base tunnel on the strength of one odd packet. + f.messageMetrics.RxInvalid(1) + if f.l.Enabled(context.Background(), slog.LevelDebug) { + hostinfo.logger(f.l).Debug("Unknown multiport lane", "from", via, "header", h) + } + return + } + if isMessageRelay { + // A relay carrier is always the base tunnel, so lane ciphertext can + // never legitimately arrive wrapped in one. + f.messageMetrics.RxInvalid(1) + if f.l.Enabled(context.Background(), slog.LevelDebug) { + hostinfo.logger(f.l).Debug("Refusing relayed multiport lane packet", "from", via, "header", h) + } + return + } + } + + if len(packet) < header.Len+ci.dKey.Overhead() { f.messageMetrics.RxInvalid(1) if f.l.Enabled(context.Background(), slog.LevelDebug) { f.l.Debug("packet too small", "from", via, "length", len(packet)) @@ -118,7 +146,7 @@ func (f *Interface) readOutsidePackets(via ViaSender, packet []byte, rxc *rxCont // All remaining packets are encrypted if isMessageRelay { // Relay packets are special, this branch should always early-return - err = hostinfo.ConnectionState.VerifyRelay(f.l, h.MessageCounter, packet, rxc.nb) + err = ci.VerifyRelay(f.l, h.MessageCounter, packet, rxc.nb) if err != nil { if f.l.Enabled(context.Background(), slog.LevelDebug) { hostinfo.logger(f.l).Debug("Failed to verify relay packet", "error", err, "from", via, "header", h) @@ -129,7 +157,7 @@ func (f *Interface) readOutsidePackets(via ViaSender, packet []byte, rxc *rxCont return } - out, err := hostinfo.ConnectionState.Decrypt(f.l, h.MessageCounter, packet, rxc.nb) + out, err := ci.Decrypt(f.l, h.MessageCounter, packet, rxc.nb) if err != nil { if f.l.Enabled(context.Background(), slog.LevelDebug) { hostinfo.logger(f.l).Debug("Failed to decrypt packet", "error", err, "from", via, "header", h) @@ -137,15 +165,19 @@ func (f *Interface) readOutsidePackets(via ViaSender, packet []byte, rxc *rxCont return } - // Roam before we respond - f.handleHostRoaming(hostinfo, via) + // Roam before we respond, but only on the base tunnel: a lane's source + // address is a per-lane 4-tuple, not the tunnel's remote, and letting it + // roam the hostinfo would point every non-lane packet at a lane port. + if lane == 0 { + f.handleHostRoaming(hostinfo, via) + } f.connectionManager.In(hostinfo) switch h.Type { case header.Message: switch h.Subtype { case header.MessageNone: - f.handleOutsideMessagePacket(hostinfo, h.MessageCounter, out, rxc) + f.handleOutsideMessagePacket(hostinfo, ci, h.MessageCounter, out, rxc) default: hostinfo.logger(f.l).Error("IsValidSubType was true, but unexpected message subtype seen", "from", via, "header", h) return @@ -170,6 +202,10 @@ func (f *Interface) readOutsidePackets(via ViaSender, packet []byte, rxc *rxCont return } f.send(header.Test, header.TestReply, hostinfo.ConnectionState, hostinfo, out, rxc.nb, rxc.scratch[:0]) + case header.LaneProbe: + f.handleLaneProbe(hostinfo, lane, out, rxc) + case header.LaneProbeAck: + f.handleLaneProbeAck(hostinfo, out) default: hostinfo.logger(f.l).Error("IsValidSubType was true, but unexpected test subtype seen", "from", via, "header", h) return @@ -471,7 +507,7 @@ func parseV4(data []byte, incoming bool, fp *firewall.ParsedPacket) error { return nil } -func (f *Interface) handleOutsideMessagePacket(hostinfo *HostInfo, messageCounter uint64, out []byte, rxc *rxContext) { +func (f *Interface) handleOutsideMessagePacket(hostinfo *HostInfo, ci *ConnectionState, messageCounter uint64, out []byte, rxc *rxContext) { err := newPacket(out, true, rxc.fwPacket) if err != nil { hostinfo.logger(f.l).Warn("Error while validating inbound packet", "error", err, "packet", out) @@ -480,6 +516,8 @@ func (f *Interface) handleOutsideMessagePacket(hostinfo *HostInfo, messageCounte dropReason := f.firewall.Drop(rxc.fwPacket.Packet, true, hostinfo, f.pki.GetCAPool(), rxc.ctCache.Get()) if dropReason != nil { + // The reject rides the base tunnel: it is a control response, not lane + // data, and the lane it arrived on says nothing about where it belongs. f.rejectOutside(out, hostinfo.ConnectionState, hostinfo, rxc.nb, rxc.scratch, rxc.q) if f.l.Enabled(context.Background(), slog.LevelDebug) { hostinfo.logger(f.l).Debug("dropping inbound packet", "fwPacket", rxc.fwPacket, "reason", dropReason) @@ -487,7 +525,7 @@ func (f *Interface) handleOutsideMessagePacket(hostinfo *HostInfo, messageCounte return } - err = f.batchers[rxc.q].Commit(out, batch.SortKey{Epoch: hostinfo.ConnectionState.epoch, Counter: messageCounter}, rxc.fwPacket) + err = f.batchers[rxc.q].Commit(out, batch.SortKey{Epoch: ci.epoch, Counter: messageCounter}, rxc.fwPacket) if err != nil { f.l.Error("Failed to write to tun", "error", err) }