From 0488793a626fa2681dc17d7f3bc44f56d8d98c7f Mon Sep 17 00:00:00 2001 From: JackDoan Date: Tue, 21 Jul 2026 10:52:24 -0500 Subject: [PATCH] crazy multiport stuff --- connection_manager.go | 87 ++++++- connection_state_test.go | 2 + control.go | 15 +- control_test.go | 2 +- examples/config.yml | 30 +++ handshake/handshake.proto | 16 +- handshake/helpers_test.go | 1 + handshake/machine.go | 34 ++- handshake/machine_lanes_test.go | 113 ++++++++ handshake/machine_test.go | 2 + handshake/payload.go | 110 ++++++++ handshake/payload_test.go | 148 ++++++++++- handshake_manager.go | 444 +++++++++++++++++++++++++++++++- hostmap.go | 203 +++++++++++++++ inside.go | 59 ++++- interface.go | 74 +++++- lanes_test.go | 409 +++++++++++++++++++++++++++++ main.go | 112 ++++++-- outside.go | 16 +- 19 files changed, 1801 insertions(+), 76 deletions(-) create mode 100644 handshake/machine_lanes_test.go create mode 100644 lanes_test.go diff --git a/connection_manager.go b/connection_manager.go index 88f31321..78f7034b 100644 --- a/connection_manager.go +++ b/connection_manager.go @@ -191,6 +191,28 @@ func (cm *connectionManager) doTrafficCheck(localIndex uint32, p, nb, out []byte } cm.resetRelayTrafficCheck(hostinfo) + cm.ensureLanes(localIndex, decision, hostinfo) +} + +// ensureLanes piggybacks lane re-establishment on the per-tunnel traffic +// tick: any live base with lane state gets its empty slots retried (subject to +// the per-slot backoff). makeTrafficDecision returns a nil hostinfo on some +// keep-alive paths, so re-resolve the index in that case — an idle base must +// still restart lanes that died while it was quiet. +func (cm *connectionManager) ensureLanes(localIndex uint32, decision trafficDecision, hostinfo *HostInfo) { + if decision == deleteTunnel || decision == closeTunnel { + return + } + if hostinfo == nil { + hostinfo = cm.hostMap.QueryIndex(localIndex) + if hostinfo == nil { + return + } + } + if hostinfo.isLane() || hostinfo.lanes == nil { + return + } + cm.intf.handshakeManager.EnsureLanes(hostinfo) } func (cm *connectionManager) resetRelayTrafficCheck(hostinfo *HostInfo) { @@ -323,6 +345,10 @@ func (cm *connectionManager) makeTrafficDecision(localIndex uint32, now time.Tim return closeTunnel, hostinfo, nil } + if hostinfo.isLane() { + return cm.makeLaneTrafficDecision(hostinfo, now) + } + primary := cm.hostMap.Hosts[hostinfo.vpnAddrs[0]] mainHostInfo := true if primary != nil && primary != hostinfo { @@ -419,13 +445,72 @@ 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.pendingDeletion.Store(false) + cm.trafficTimer.Add(hostinfo.localIndexId, cm.checkInterval) + return doNothing, hostinfo, nil + } + + if hostinfo.pendingDeletion.Load() { + 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.pendingDeletion.Store(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 } - inactiveDuration := now.Sub(hostinfo.lastUsed) + // 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) if inactiveDuration < cm.getInactivityTimeout() { // It's not considered inactive return inactiveDuration, false diff --git a/connection_state_test.go b/connection_state_test.go index dea60d39..fc841f6e 100644 --- a/connection_state_test.go +++ b/connection_state_test.go @@ -55,6 +55,7 @@ func runTestHandshake(t *testing.T) (initR, respR *handshake.Result) { cert.Version2, initCreds, verifier, func() (uint32, error) { return 1000, nil }, true, header.HandshakeIXPSK0, + nil, ) require.NoError(t, err) @@ -62,6 +63,7 @@ func runTestHandshake(t *testing.T) (initR, respR *handshake.Result) { cert.Version2, respCreds, verifier, func() (uint32, error) { return 2000, nil }, false, header.HandshakeIXPSK0, + nil, ) require.NoError(t, err) diff --git a/control.go b/control.go index a79ebbfa..299dfe26 100644 --- a/control.go +++ b/control.go @@ -66,6 +66,9 @@ 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"` } // Start actually runs nebula, this is a nonblocking call. @@ -198,7 +201,9 @@ func (c *Control) RebindUDPServer() { return } - _ = c.f.outside.Rebind() + for _, w := range c.f.writers { + _ = w.Rebind() + } // Trigger a lighthouse update, useful for mobile clients that should have an update interval of 0 c.f.lightHouse.SendUpdate() @@ -349,6 +354,11 @@ 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) } @@ -377,6 +387,9 @@ 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, } for i, a := range h.vpnAddrs { diff --git a/control_test.go b/control_test.go index 94ee4ee3..fb0a79ed 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"}, thi) + assertFields(t, []string{"VpnAddrs", "LocalIndex", "RemoteIndex", "RemoteAddrs", "Cert", "MessageCounter", "CurrentRemote", "CurrentRelaysToMe", "CurrentRelaysThroughMe", "IsLane", "LaneIndex", "SockIdx"}, thi) assert.Equal(t, &expectedInfo, thi) test.AssertDeepCopyEqual(t, &expectedInfo, thi) diff --git a/examples/config.yml b/examples/config.yml index 4c64f7ab..6b1bb3d0 100644 --- a/examples/config.yml +++ b/examples/config.yml @@ -169,6 +169,36 @@ listen: # This option is only supported on Linux. #routines: 1 +# EXPERIMENTAL: multiport lanes give each pair of hosts multiple underlay UDP +# flows so overlay traffic is no longer bottlenecked by a single 5-tuple +# (one ECMP path, one NIC RSS queue, one per-flow policer). Socket i binds +# listen.port+i instead of sharing one port via SO_REUSEPORT, and one extra +# tunnel ("lane") per routine is negotiated with capable peers: lane i +# handshakes from local port listen.port+i to the peer's advertised +# base+(i mod peer_ports). Each lane is a full Noise session with its own +# keys, nonce counter and replay window, so flows taking different paths +# never fight over shared replay state. +# +# Peers negotiate lanes in the handshake; vanilla peers get a single normal +# tunnel. All control traffic (handshakes, lighthouse, punching, relays) and +# the data fallback stay on the base tunnel/port. Lanes are established after +# the base tunnel comes up, are kept alive with their own keepalives, and +# traffic falls back to the base tunnel while a lane is down. +# +# Requirements: routines > 1, Linux, and the port range +# [listen.port, listen.port+routines-1] reachable through firewalls on both +# sides (peers behind NAT fall back to the base tunnel). With listen.port 0 +# the base port is dynamic and the next routines-1 ports above it are +# claimed. Enabled by default when the requirements hold; degrades to a +# single port otherwise. Not reloadable. +#multiport: + # Bind `routines` consecutive UDP ports and negotiate lanes with peers. + #enabled: true + # How many lanes to run, counting the base tunnel as lane 0. 0 (default) + # means one per routine. Lowering this bounds how many extra tunnels each + # peer pair maintains; routines without a lane use the base tunnel. + #lanes: 0 + punchy: # Continues to punch inbound/outbound at a regular interval to avoid expiration of firewall nat mappings # This setting is reloadable. diff --git a/handshake/handshake.proto b/handshake/handshake.proto index 8eb32aa6..42f1ed62 100644 --- a/handshake/handshake.proto +++ b/handshake/handshake.proto @@ -23,7 +23,19 @@ message NebulaHandshakeDetails { // hand-written parser silently skips it on read. uint64 Cookie = 4 [deprecated = true]; uint64 Time = 5; + // Multiport lane negotiation. Absent on hosts without multiport enabled; + // vanilla nebula treats 6 and 7 as unknown fields and skips them. + LaneDetails InitiatorLanes = 6; + LaneDetails ResponderLanes = 7; uint32 CertVersion = 8; - // reserved for WIP multiport - reserved 6, 7; +} + +// LaneDetails advertises a host's multiport lane capability. On a base +// handshake LaneIndex is 0 and PortCount/BasePort describe the sender's +// consecutively bound UDP ports. On a lane handshake the initiator sets +// LaneIndex to its (nonzero) lane number. +message LaneDetails { + uint32 PortCount = 1; + uint32 BasePort = 2; + uint32 LaneIndex = 3; } diff --git a/handshake/helpers_test.go b/handshake/helpers_test.go index c72346cb..7ba6481d 100644 --- a/handshake/helpers_test.go +++ b/handshake/helpers_test.go @@ -71,6 +71,7 @@ func newTestMachine( cs.version, cs.getCredential, verifier, func() (uint32, error) { return localIndex, nil }, initiator, header.HandshakeIXPSK0, + nil, ) require.NoError(t, err) return m diff --git a/handshake/machine.go b/handshake/machine.go index baf61589..50b29bdc 100644 --- a/handshake/machine.go +++ b/handshake/machine.go @@ -39,6 +39,13 @@ type Result struct { HandshakeTime uint64 MessageIndex uint64 // number of messages exchanged during the handshake Initiator bool + + // 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. + PeerPortCount uint32 + PeerBasePort uint32 + PeerLaneIndex uint32 } // Machine drives a Noise handshake through N messages. It handles Noise @@ -61,6 +68,7 @@ type Machine struct { verifier CertVerifier result *Result msgs []msgFlags + lanes *LaneDetails // our multiport advert; nil emits a vanilla payload myVersion cert.Version subtype header.MessageSubType indexAllocated bool @@ -73,6 +81,8 @@ type Machine struct { // the noise pattern and the per-message content layout. The credential for // `version` is fetched via getCred and used to seed the noise.HandshakeState. // IndexAllocator is called lazily when the first outgoing payload is built. +// lanes, when non-nil, is emitted as this side's multiport advert on every +// payload-bearing message; nil produces byte-identical vanilla payloads. func NewMachine( version cert.Version, getCred GetCredentialFunc, @@ -80,6 +90,7 @@ func NewMachine( allocIndex IndexAllocator, initiator bool, subtype header.MessageSubType, + lanes *LaneDetails, ) (*Machine, error) { info, err := subtypeInfoFor(subtype) if err != nil { @@ -103,6 +114,7 @@ func NewMachine( getCred: getCred, allocIndex: allocIndex, verifier: verifier, + lanes: lanes, myVersion: version, result: &Result{ Initiator: initiator, @@ -298,7 +310,8 @@ func (m *Machine) processPayload(msg []byte, flags msgFlags) error { } // Assert the payload contains exactly what we expect - hasPayloadData := payload.InitiatorIndex != 0 || payload.ResponderIndex != 0 || payload.Time != 0 + hasPayloadData := payload.InitiatorIndex != 0 || payload.ResponderIndex != 0 || payload.Time != 0 || + payload.InitiatorLanes != nil || payload.ResponderLanes != nil if hasPayloadData != flags.expectsPayload { m.failed = true return ErrUnexpectedContent @@ -327,6 +340,23 @@ func (m *Machine) processPayload(msg []byte, flags msgFlags) error { m.result.RemoteIndex = remoteIndex m.result.HandshakeTime = payload.Time m.payloadSet = true + + // 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. + var peerLanes *LaneDetails + if m.result.Initiator { + peerLanes = payload.ResponderLanes + } else { + peerLanes = payload.InitiatorLanes + } + if peerLanes != nil && peerLanes.BasePort <= 0xffff && peerLanes.PortCount <= 0xffff { + m.result.PeerPortCount = peerLanes.PortCount + m.result.PeerBasePort = peerLanes.BasePort + m.result.PeerLaneIndex = peerLanes.LaneIndex + } } // Process certificate @@ -397,9 +427,11 @@ func (m *Machine) marshalOutgoing(flags msgFlags) ([]byte, error) { if m.result.Initiator { p.InitiatorIndex = m.result.LocalIndex + p.InitiatorLanes = m.lanes } else { p.ResponderIndex = m.result.LocalIndex p.InitiatorIndex = m.result.RemoteIndex + p.ResponderLanes = m.lanes } p.Time = uint64(time.Now().UnixNano()) } diff --git a/handshake/machine_lanes_test.go b/handshake/machine_lanes_test.go new file mode 100644 index 00000000..ef6c0c6b --- /dev/null +++ b/handshake/machine_lanes_test.go @@ -0,0 +1,113 @@ +package handshake + +import ( + "net/netip" + "testing" + "time" + + "github.com/slackhq/nebula/cert" + ct "github.com/slackhq/nebula/cert_test" + "github.com/slackhq/nebula/header" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// newTestLaneMachine is newTestMachine with a lane advert attached. +func newTestLaneMachine( + t *testing.T, + cs *testCertState, + verifier CertVerifier, + initiator bool, + localIndex uint32, + lanes *LaneDetails, +) *Machine { + t.Helper() + m, err := NewMachine( + cs.version, cs.getCredential, + verifier, func() (uint32, error) { return localIndex, nil }, + initiator, header.HandshakeIXPSK0, + lanes, + ) + require.NoError(t, err) + return m +} + +func doFullLaneHandshake(t *testing.T, initLanes, respLanes *LaneDetails) (initR, respR *Result) { + t.Helper() + ca, _, caKey, _ := ct.NewTestCaCert( + cert.Version2, cert.Curve_CURVE25519, time.Time{}, time.Time{}, nil, nil, nil, + ) + caPool := ct.NewTestCAPool(ca) + v := testVerifier(caPool) + + initCS := newTestCertState(t, ca, caKey, "initiator", []netip.Prefix{netip.MustParsePrefix("10.0.0.1/24")}) + respCS := newTestCertState(t, ca, caKey, "responder", []netip.Prefix{netip.MustParsePrefix("10.0.0.2/24")}) + + initM := newTestLaneMachine(t, initCS, v, true, 1000, initLanes) + respM := newTestLaneMachine(t, respCS, v, false, 2000, respLanes) + + msg1, err := initM.Initiate(nil) + require.NoError(t, err) + + resp, respR, err := respM.ProcessPacket(nil, msg1) + require.NoError(t, err) + require.NotNil(t, respR) + + _, initR, err = initM.ProcessPacket(nil, resp) + require.NoError(t, err) + require.NotNil(t, initR) + return initR, respR +} + +func TestMachineLaneAdvertBothSides(t *testing.T) { + initR, respR := doFullLaneHandshake(t, + &LaneDetails{PortCount: 8, BasePort: 4242}, + &LaneDetails{PortCount: 4, BasePort: 5353}, + ) + + // 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(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) +} + +func TestMachineLaneAdvertAsymmetric(t *testing.T) { + // Vanilla initiator, multiport responder and vice versa: the nil side + // yields all-zero peer fields on the other end. + initR, respR := doFullLaneHandshake(t, nil, &LaneDetails{PortCount: 4, BasePort: 5353}) + assert.Equal(t, uint32(4), initR.PeerPortCount) + assert.Equal(t, uint32(0), respR.PeerPortCount) + assert.Equal(t, uint32(0), respR.PeerBasePort) + + initR, respR = doFullLaneHandshake(t, &LaneDetails{PortCount: 8, BasePort: 4242}, nil) + assert.Equal(t, uint32(0), initR.PeerPortCount) + assert.Equal(t, uint32(8), respR.PeerPortCount) +} + +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}, + ) + assert.Equal(t, uint32(0), respR.PeerPortCount) + assert.Equal(t, uint32(0), respR.PeerBasePort) + // The sane side still negotiates. + assert.Equal(t, uint32(4), initR.PeerPortCount) +} diff --git a/handshake/machine_test.go b/handshake/machine_test.go index 01c968ed..7c97950e 100644 --- a/handshake/machine_test.go +++ b/handshake/machine_test.go @@ -444,6 +444,7 @@ func TestMachineThreeMessagePattern(t *testing.T) { initCS.getCredential, v, func() (uint32, error) { return 1000, nil }, true, header.HandshakeXXPSK0, + nil, ) require.NoError(t, err) @@ -452,6 +453,7 @@ func TestMachineThreeMessagePattern(t *testing.T) { respCS.getCredential, v, func() (uint32, error) { return 2000, nil }, false, header.HandshakeXXPSK0, + nil, ) require.NoError(t, err) diff --git a/handshake/payload.go b/handshake/payload.go index 4567fc0d..0e8a6c37 100644 --- a/handshake/payload.go +++ b/handshake/payload.go @@ -20,6 +20,19 @@ type Payload struct { ResponderIndex uint32 Time uint64 CertVersion uint32 + + // Multiport lane negotiation; nil when the sender has multiport disabled + // (which keeps the encoded payload byte-identical to a vanilla one). + InitiatorLanes *LaneDetails + ResponderLanes *LaneDetails +} + +// LaneDetails advertises multiport lane capability. LaneIndex is zero on base +// handshakes and the initiator's lane number (>= 1) on lane handshakes. +type LaneDetails struct { + PortCount uint32 + BasePort uint32 + LaneIndex uint32 } // Proto field numbers for NebulaHandshakeDetails @@ -28,9 +41,18 @@ const ( fieldInitiatorIndex = 2 // uint32 fieldResponderIndex = 3 // uint32 fieldTime = 5 // uint64 + fieldInitiatorLanes = 6 // LaneDetails + fieldResponderLanes = 7 // LaneDetails fieldCertVersion = 8 // uint32 ) +// Proto field numbers for LaneDetails +const ( + fieldLanePortCount = 1 // uint32 + fieldLaneBasePort = 2 // uint32 + fieldLaneLaneIndex = 3 // uint32 +) + // MarshalPayload encodes a handshake payload in protobuf wire format compatible // with NebulaHandshake{Details: NebulaHandshakeDetails{...}}. // Returns out (which may be nil), with the marshalled Payload appended to it. @@ -53,6 +75,14 @@ func MarshalPayload(out []byte, p Payload) []byte { details = protowire.AppendTag(details, fieldTime, protowire.VarintType) details = protowire.AppendVarint(details, p.Time) } + if p.InitiatorLanes != nil { + details = protowire.AppendTag(details, fieldInitiatorLanes, protowire.BytesType) + details = protowire.AppendBytes(details, p.InitiatorLanes.marshal(nil)) + } + if p.ResponderLanes != nil { + details = protowire.AppendTag(details, fieldResponderLanes, protowire.BytesType) + details = protowire.AppendBytes(details, p.ResponderLanes.marshal(nil)) + } if p.CertVersion != 0 { details = protowire.AppendTag(details, fieldCertVersion, protowire.VarintType) details = protowire.AppendVarint(details, uint64(p.CertVersion)) @@ -64,6 +94,20 @@ func MarshalPayload(out []byte, p Payload) []byte { return out } +// marshal appends the LaneDetails submessage fields to out. All fields are +// emitted unconditionally: a LaneDetails is only present at all when multiport +// is negotiating, and explicit zeros keep the parser's presence semantics +// trivial. +func (d *LaneDetails) marshal(out []byte) []byte { + out = protowire.AppendTag(out, fieldLanePortCount, protowire.VarintType) + 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)) + return out +} + // UnmarshalPayload decodes a protobuf-encoded NebulaHandshake message. func UnmarshalPayload(b []byte) (Payload, error) { var p Payload @@ -161,6 +205,72 @@ func unmarshalPayloadDetails(p *Payload, b []byte) error { } p.CertVersion = uint32(v) b = b[n:] + case fieldInitiatorLanes: + if typ != protowire.BytesType { + return errInvalidHandshakeDetails + } + v, n := protowire.ConsumeBytes(b) + if n < 0 { + return errInvalidHandshakeDetails + } + p.InitiatorLanes = new(LaneDetails) + if err := unmarshalLaneDetails(p.InitiatorLanes, v); err != nil { + return err + } + b = b[n:] + case fieldResponderLanes: + if typ != protowire.BytesType { + return errInvalidHandshakeDetails + } + v, n := protowire.ConsumeBytes(b) + if n < 0 { + return errInvalidHandshakeDetails + } + p.ResponderLanes = new(LaneDetails) + if err := unmarshalLaneDetails(p.ResponderLanes, v); err != nil { + return err + } + b = b[n:] + default: + n := protowire.ConsumeFieldValue(num, typ, b) + if n < 0 { + return errInvalidHandshakeDetails + } + b = b[n:] + } + } + return nil +} + +func unmarshalLaneDetails(d *LaneDetails, b []byte) error { + for len(b) > 0 { + num, typ, n := protowire.ConsumeTag(b) + if n < 0 { + return errInvalidHandshakeDetails + } + b = b[n:] + + // Same contract as the details parser: known fields hard-fail on a + // wire-type mismatch, unknown fields are skipped, repeated singular + // fields follow proto3 last-wins. + switch num { + case fieldLanePortCount, fieldLaneBasePort, fieldLaneLaneIndex: + if typ != protowire.VarintType { + return errInvalidHandshakeDetails + } + v, n := protowire.ConsumeVarint(b) + if n < 0 || v > math.MaxUint32 { + return errInvalidHandshakeDetails + } + switch num { + case fieldLanePortCount: + d.PortCount = uint32(v) + case fieldLaneBasePort: + d.BasePort = uint32(v) + case fieldLaneLaneIndex: + d.LaneIndex = uint32(v) + } + b = b[n:] default: n := protowire.ConsumeFieldValue(num, typ, b) if n < 0 { diff --git a/handshake/payload_test.go b/handshake/payload_test.go index 2ff3231c..73899c64 100644 --- a/handshake/payload_test.go +++ b/handshake/payload_test.go @@ -117,23 +117,134 @@ func TestPayloadUnknownFields(t *testing.T) { assert.Equal(t, uint32(88), got.ResponderIndex) }) - t.Run("reserved fields 6 and 7 are skipped", func(t *testing.T) { - // Fields 6 and 7 are reserved in the proto definition + t.Run("unknown field inside LaneDetails is skipped", func(t *testing.T) { + var lane []byte + lane = protowire.AppendTag(lane, fieldLanePortCount, protowire.VarintType) + lane = protowire.AppendVarint(lane, 4) + lane = protowire.AppendTag(lane, 50, protowire.VarintType) // unknown subfield + lane = protowire.AppendVarint(lane, 9999) + lane = protowire.AppendTag(lane, fieldLaneBasePort, protowire.VarintType) + lane = protowire.AppendVarint(lane, 4242) + var details []byte details = protowire.AppendTag(details, fieldInitiatorIndex, protowire.VarintType) details = protowire.AppendVarint(details, 100) - details = protowire.AppendTag(details, 6, protowire.VarintType) - details = protowire.AppendVarint(details, 1) - details = protowire.AppendTag(details, 7, protowire.VarintType) - details = protowire.AppendVarint(details, 2) + details = protowire.AppendTag(details, fieldInitiatorLanes, protowire.BytesType) + details = protowire.AppendBytes(details, lane) - var data []byte - data = protowire.AppendTag(data, 1, protowire.BytesType) - data = protowire.AppendBytes(data, details) + got, err := UnmarshalPayload(wrapDetails(details)) + require.NoError(t, err) + assert.Equal(t, uint32(100), got.InitiatorIndex) + require.NotNil(t, got.InitiatorLanes) + assert.Equal(t, uint32(4), got.InitiatorLanes.PortCount) + assert.Equal(t, uint32(4242), got.InitiatorLanes.BasePort) + }) +} + +func TestPayloadLaneDetails(t *testing.T) { + t.Run("round trip both sides", func(t *testing.T) { + data := MarshalPayload(nil, Payload{ + InitiatorIndex: 12345, + Time: 999, + InitiatorLanes: &LaneDetails{PortCount: 8, BasePort: 4242, LaneIndex: 3}, + ResponderLanes: &LaneDetails{PortCount: 4, BasePort: 5353}, + }) got, err := UnmarshalPayload(data) require.NoError(t, err) - assert.Equal(t, uint32(100), got.InitiatorIndex) + require.NotNil(t, got.InitiatorLanes) + assert.Equal(t, LaneDetails{PortCount: 8, BasePort: 4242, LaneIndex: 3}, *got.InitiatorLanes) + require.NotNil(t, got.ResponderLanes) + assert.Equal(t, LaneDetails{PortCount: 4, BasePort: 5353}, *got.ResponderLanes) + }) + + t.Run("zero-valued LaneDetails survives the round trip", func(t *testing.T) { + // Presence is what negotiation keys on; an all-zero advert must not + // decay to nil. + data := MarshalPayload(nil, Payload{ + InitiatorIndex: 1, + InitiatorLanes: &LaneDetails{}, + }) + got, err := UnmarshalPayload(data) + require.NoError(t, err) + require.NotNil(t, got.InitiatorLanes) + assert.Equal(t, LaneDetails{}, *got.InitiatorLanes) + assert.Nil(t, got.ResponderLanes) + }) + + t.Run("nil lanes marshal byte-identical to a vanilla payload", func(t *testing.T) { + p := Payload{ + Cert: []byte("cert"), + CertVersion: 2, + InitiatorIndex: 100, + Time: 999, + } + // The vanilla encoding of the same fields, built by hand in field order. + var details []byte + details = protowire.AppendTag(details, fieldCert, protowire.BytesType) + details = protowire.AppendBytes(details, p.Cert) + details = protowire.AppendTag(details, fieldInitiatorIndex, protowire.VarintType) + details = protowire.AppendVarint(details, uint64(p.InitiatorIndex)) + details = protowire.AppendTag(details, fieldTime, protowire.VarintType) + details = protowire.AppendVarint(details, p.Time) + details = protowire.AppendTag(details, fieldCertVersion, protowire.VarintType) + details = protowire.AppendVarint(details, uint64(p.CertVersion)) + + assert.Equal(t, wrapDetails(details), MarshalPayload(nil, p)) + }) + + t.Run("lane field with wrong wire type rejected", func(t *testing.T) { + for _, field := range []protowire.Number{fieldInitiatorLanes, fieldResponderLanes} { + var details []byte + details = protowire.AppendTag(details, field, protowire.VarintType) + details = protowire.AppendVarint(details, 1) + _, err := UnmarshalPayload(wrapDetails(details)) + assert.Error(t, err) + } + }) + + t.Run("lane subfield with wrong wire type rejected", func(t *testing.T) { + var lane []byte + lane = protowire.AppendTag(lane, fieldLanePortCount, protowire.BytesType) + lane = protowire.AppendBytes(lane, []byte{1, 2, 3}) + + var details []byte + details = protowire.AppendTag(details, fieldInitiatorLanes, protowire.BytesType) + details = protowire.AppendBytes(details, lane) + _, err := UnmarshalPayload(wrapDetails(details)) + assert.Error(t, err) + }) + + t.Run("truncated LaneDetails submessage rejected", func(t *testing.T) { + var details []byte + details = protowire.AppendTag(details, fieldInitiatorLanes, protowire.BytesType) + details = append(details, 0x0a, 0x01, 0x02) // length 10, only 2 bytes + _, err := UnmarshalPayload(wrapDetails(details)) + assert.Error(t, err) + }) + + t.Run("truncated varint inside LaneDetails rejected", func(t *testing.T) { + var lane []byte + lane = protowire.AppendTag(lane, fieldLaneBasePort, protowire.VarintType) + lane = append(lane, 0x80) // incomplete varint + + var details []byte + details = protowire.AppendTag(details, fieldResponderLanes, protowire.BytesType) + details = protowire.AppendBytes(details, lane) + _, err := UnmarshalPayload(wrapDetails(details)) + assert.Error(t, err) + }) + + t.Run("lane subfield varint overflow rejected", func(t *testing.T) { + var lane []byte + lane = protowire.AppendTag(lane, fieldLaneLaneIndex, protowire.VarintType) + lane = protowire.AppendVarint(lane, math.MaxUint32+1) + + var details []byte + details = protowire.AppendTag(details, fieldInitiatorLanes, protowire.BytesType) + details = protowire.AppendBytes(details, lane) + _, err := UnmarshalPayload(wrapDetails(details)) + assert.Error(t, err) }) } @@ -328,6 +439,12 @@ func FuzzPayload(f *testing.F) { Time: 3, CertVersion: 2, })) + f.Add(MarshalPayload(nil, Payload{ + InitiatorIndex: 1, + Time: 3, + InitiatorLanes: &LaneDetails{PortCount: 8, BasePort: 4242, LaneIndex: 2}, + ResponderLanes: &LaneDetails{PortCount: 4, BasePort: 5353}, + })) f.Add([]byte{}) f.Add([]byte{0xff}) @@ -357,5 +474,14 @@ func payloadsEqual(a, b Payload) bool { a.InitiatorIndex == b.InitiatorIndex && a.ResponderIndex == b.ResponderIndex && a.Time == b.Time && - a.CertVersion == b.CertVersion + a.CertVersion == b.CertVersion && + laneDetailsEqual(a.InitiatorLanes, b.InitiatorLanes) && + laneDetailsEqual(a.ResponderLanes, b.ResponderLanes) +} + +func laneDetailsEqual(a, b *LaneDetails) bool { + if a == nil || b == nil { + return a == b + } + return *a == *b } diff --git a/handshake_manager.go b/handshake_manager.go index 99d5a72c..aefc866a 100644 --- a/handshake_manager.go +++ b/handshake_manager.go @@ -50,6 +50,14 @@ type HandshakeConfig struct { retries int64 triggerBuffer int + // 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 + // own bound port range and are advertised in every handshake payload. + laneCount int + lanePortCount uint16 + laneBasePort uint16 + messageMetrics *MessageMetrics } @@ -65,11 +73,15 @@ type HandshakeManager struct { outside udp.Conn config HandshakeConfig OutboundHandshakeTimer *LockingTimerWheel[netip.Addr] - messageMetrics *MessageMetrics - metricInitiated metrics.Counter - metricTimedOut metrics.Counter - f *Interface - l *slog.Logger + // 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 // can be used to trigger outbound handshake for the given vpnIp trigger chan netip.Addr @@ -88,6 +100,11 @@ 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) { @@ -125,6 +142,7 @@ 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), @@ -144,6 +162,7 @@ func (hm *HandshakeManager) Run(ctx context.Context) { hm.handleOutbound(vpnIP, true) case now := <-clockSource.C: hm.NextOutboundHandshakeTimerTick(now) + hm.NextOutboundLaneTimerTick(now) } } } @@ -529,7 +548,13 @@ func (hm *HandshakeManager) DeleteHostInfo(hostinfo *HostInfo) { func (hm *HandshakeManager) unlockedDeleteHostInfo(hostinfo *HostInfo) { for _, addr := range hostinfo.vpnAddrs { - delete(hm.vpnIps, addr) + // 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. + if cur, ok := hm.vpnIps[addr]; ok && cur.hostinfo == hostinfo { + delete(hm.vpnIps, addr) + } } if len(hm.vpnIps) == 0 { @@ -664,6 +689,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)), ) if err != nil { hm.f.l.Error("Failed to create handshake machine", @@ -687,6 +713,215 @@ func (hm *HandshakeManager) buildStage0Packet(hh *HandshakeHostInfo) bool { return true } +// 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 { + if hm.config.laneCount == 0 { + return nil + } + return &handshake.LaneDetails{ + PortCount: uint32(hm.config.lanePortCount), + BasePort: uint32(hm.config.laneBasePort), + LaneIndex: laneIndex, + } +} + +// 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 + } + hostinfo.lanes = newLaneState(hm.f.routines, uint16(peerPorts), uint16(result.PeerBasePort)) +} + +// EnsureLanes starts lane handshakes for every empty, non-pending, retry-due +// slot of a base tunnel. Called on base handshake completion (both sides) and +// from the connection manager's per-tunnel tick, so a dead lane re-establishes +// within one check interval, subject to per-slot backoff. +func (hm *HandshakeManager) EnsureLanes(base *HostInfo) { + ls := base.lanes + if ls == nil || hm.config.laneCount <= 1 { + 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 + } + 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.peerBasePort+uint16(i%int(ls.peerPortCount))) + + 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) + 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, + ) + } + + 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) +} + // beginHandshake handles an incoming handshake packet that doesn't match any // existing pending handshake. It creates a new responder Machine and processes // the first message. @@ -705,6 +940,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), ) if err != nil { f.l.Error("Failed to create handshake machine", "from", via, "error", err) @@ -741,6 +977,13 @@ 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 + } + hostinfo := &HostInfo{ ConnectionState: newConnectionStateFromResult(result), localIndexId: result.LocalIndex, @@ -785,6 +1028,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) existing, err := hm.CheckAndComplete(hostinfo, handshakePacketStage0, f) if err != nil { @@ -794,6 +1038,7 @@ 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) { @@ -801,6 +1046,139 @@ 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) + } + + hostinfo := &HostInfo{ + ConnectionState: newConnectionStateFromResult(result), + 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 @@ -821,6 +1199,14 @@ 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", @@ -833,7 +1219,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.DeleteHostInfo(hostinfo) + hm.deletePendingHostInfo(hostinfo) return } @@ -843,7 +1229,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.DeleteHostInfo(hostinfo) + hm.deletePendingHostInfo(hostinfo) } else { f.l.Debug("Failed to process handshake packet", "vpnAddrs", hostinfo.vpnAddrs, "from", via, "error", err) @@ -866,7 +1252,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.DeleteHostInfo(hostinfo) + hm.deletePendingHostInfo(hostinfo) return } @@ -900,7 +1286,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.DeleteHostInfo(hostinfo) + hm.deletePendingHostInfo(hostinfo) return } vpnAddrs[i] = network.Addr() @@ -924,6 +1310,15 @@ 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 @@ -958,7 +1353,31 @@ 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.Complete(hostinfo, f) + hm.EnsureLanes(hostinfo) if len(hh.packetStore) > 0 { if f.l.Enabled(context.Background(), slog.LevelDebug) { @@ -1064,7 +1483,10 @@ func (hm *HandshakeManager) sendHandshakeResponse(via ViaSender, msg []byte, hos if !via.IsRelayed { fields := append(logFields, "from", via) - err := f.outside.WriteTo(msg, via.UdpAddr) + // Reply from the socket the handshake arrived on so the initiator sees + // the source port it targeted. Identical to f.outside under vanilla + // config (all writers share one port); required for multiport lanes. + err := f.writers[via.SockIdx].WriteTo(msg, via.UdpAddr) if err != nil { f.l.Error("Failed to send handshake message", append(fields, "error", err)...) } else { diff --git a/hostmap.go b/hostmap.go index 45515fc3..9ac6e38b 100644 --- a/hostmap.go +++ b/hostmap.go @@ -282,6 +282,135 @@ type HostInfo struct { // This value will be behind against actual tunnel utilization in the hot path. // 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; lane i targets peerBasePort + (i % peerPortCount). + peerPortCount uint16 + peerBasePort 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] + + // 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 uint16) *laneState { + return &laneState{ + peerPortCount: peerPortCount, + peerBasePort: peerBasePort, + txLanes: make([]atomic.Pointer[HostInfo], laneCount), + txPending: make([]bool, laneCount), + txFails: make([]uint8, laneCount), + txRetryAt: make([]time.Time, laneCount), + } +} + +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 } type ViaSender struct { @@ -289,6 +418,11 @@ type ViaSender struct { relayHI *HostInfo // relayHI is the host info object of the relay relay *Relay // relay contains the rest of the relay information, including the PeerIP of the host trying to communicate with us. IsRelayed bool // IsRelayed is true if the packet was sent through a relay + + // SockIdx is the local socket (Interface.writers index) the packet + // arrived on. Replies that must originate from the same 4-tuple egress + // f.writers[SockIdx]. + SockIdx int } func (v ViaSender) String() string { @@ -450,6 +584,12 @@ 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 @@ -478,6 +618,19 @@ 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) + } + } + // 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. final := true @@ -543,6 +696,35 @@ func (hm *HostMap) unlockedDeleteHostInfo(hostinfo *HostInfo) bool { return final } +// 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 { @@ -671,6 +853,27 @@ 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.out.Store(true) + 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 1c08f5fb..49b92305 100644 --- a/inside.go +++ b/inside.go @@ -10,12 +10,11 @@ import ( "github.com/slackhq/nebula/header" "github.com/slackhq/nebula/iputil" "github.com/slackhq/nebula/noiseutil" - "github.com/slackhq/nebula/overlay/batch" "github.com/slackhq/nebula/overlay/tio" "github.com/slackhq/nebula/routing" ) -func (f *Interface) consumeInsidePacket(pkt tio.Packet, fwPacket *firewall.Packet, nb []byte, sendBatch batch.TxBatcher, rejectBuf []byte, q int, localCache firewall.ConntrackCache) { +func (f *Interface) consumeInsidePacket(pkt tio.Packet, fwPacket *firewall.Packet, nb []byte, tx *txQueue, rejectBuf []byte, q int, localCache firewall.ConntrackCache) { // borrowed: pkt.Bytes is owned by the originating tio.Queue and is // only valid until the next Read on that queue. Every consumer below // (parse, self-forward, handshake cache, sendInsideMessage) reads it @@ -107,7 +106,7 @@ func (f *Interface) consumeInsidePacket(pkt tio.Packet, fwPacket *firewall.Packe dropReason := f.firewall.Drop(*fwPacket, false, hostinfo, f.pki.GetCAPool(), localCache) if dropReason == nil { - f.sendInsideMessage(hostinfo, pkt, nb, sendBatch, rejectBuf, q) + f.sendInsideMessage(hostinfo, pkt, nb, tx, q) } else { f.rejectInside(packet, rejectBuf, q) if f.l.Enabled(context.Background(), slog.LevelDebug) { @@ -153,12 +152,18 @@ 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. -func (f *Interface) sendInsideMessage(hostinfo *HostInfo, pkt tio.Packet, nb []byte, sendBatch batch.TxBatcher, rejectBuf []byte, q int) { +// +// 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 +// tx.base (socket 0). +func (f *Interface) sendInsideMessage(hostinfo *HostInfo, pkt tio.Packet, nb []byte, tx *txQueue, q int) { ci := hostinfo.ConnectionState if ci.eKey == nil { return } + sendBatch := tx.base remote := hostinfo.GetRemote() ecnEnabled := f.ecnEnabled.Load() if hostinfo.lastRebindCount != f.rebindCount { @@ -224,6 +229,21 @@ 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. + if ls := hostinfo.lanes; ls != nil && q < len(ls.txLanes) { + if lane := ls.txLanes[q].Load(); lane != nil { + if lci := lane.ConnectionState; lci != nil && lci.eKey != nil { + hostinfo = lane + ci = lci + remote = lane.GetRemote() + 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) @@ -279,7 +299,7 @@ func (f *Interface) rejectInside(packet []byte, out []byte, q int) { } } -func (f *Interface) rejectOutside(packet []byte, ci *ConnectionState, hostinfo *HostInfo, nb, rejectBuf []byte, q int) { +func (f *Interface) rejectOutside(packet []byte, ci *ConnectionState, hostinfo *HostInfo, nb, rejectBuf []byte) { if !f.firewall.InboundSendReject { return } @@ -302,7 +322,7 @@ func (f *Interface) rejectOutside(packet []byte, ci *ConnectionState, hostinfo * return } - f.sendNoMetrics(header.Message, 0, ci, hostinfo, netip.AddrPort{}, out, nb, encryptBuf, q) + f.sendNoMetrics(header.Message, 0, ci, hostinfo, netip.AddrPort{}, out, nb, encryptBuf) } // Handshake will attempt to initiate a tunnel with the provided vpn address. This is a no-op if the tunnel is already established or being established @@ -415,7 +435,7 @@ func (f *Interface) sendMessageNow(t header.MessageType, st header.MessageSubTyp return } - f.sendNoMetrics(header.Message, st, hostinfo.ConnectionState, hostinfo, netip.AddrPort{}, p, nb, out, 0) + f.sendNoMetrics(header.Message, st, hostinfo.ConnectionState, hostinfo, netip.AddrPort{}, p, nb, out) } // SendMessageToVpnAddr handles real addr:port lookup and sends to the current best known address for vpnAddr. @@ -447,12 +467,12 @@ func (f *Interface) SendMessageToHostInfo(t header.MessageType, st header.Messag func (f *Interface) send(t header.MessageType, st header.MessageSubType, ci *ConnectionState, hostinfo *HostInfo, p, nb, out []byte) { f.messageMetrics.Tx(t, st, 1) - f.sendNoMetrics(t, st, ci, hostinfo, netip.AddrPort{}, p, nb, out, 0) + f.sendNoMetrics(t, st, ci, hostinfo, netip.AddrPort{}, p, nb, out) } func (f *Interface) sendTo(t header.MessageType, st header.MessageSubType, ci *ConnectionState, hostinfo *HostInfo, remote netip.AddrPort, p, nb, out []byte) { f.messageMetrics.Tx(t, st, 1) - f.sendNoMetrics(t, st, ci, hostinfo, remote, p, nb, out, 0) + f.sendNoMetrics(t, st, ci, hostinfo, remote, p, nb, out) } func (f *Interface) prepareSendVia(via *HostInfo, @@ -530,16 +550,23 @@ func (f *Interface) SendVia(via *HostInfo, return } - err = f.writers[0].WriteTo(toSend, via.GetRemote()) + // Relay carriers are base tunnels (sockIdx 0); indexing through the + // carrier keeps the invariant explicit. + err = f.writers[via.sockIdx].WriteTo(toSend, via.GetRemote()) if err != nil { via.logger(f.l).Info("Failed to WriteTo in sendVia", "error", err) } } -func (f *Interface) sendNoMetrics(t header.MessageType, st header.MessageSubType, ci *ConnectionState, hostinfo *HostInfo, remote netip.AddrPort, p, nb, out []byte, q int) { +func (f *Interface) sendNoMetrics(t header.MessageType, st header.MessageSubType, ci *ConnectionState, hostinfo *HostInfo, remote netip.AddrPort, p, nb, out []byte) { if ci.eKey == nil { return } + // Every packet on a tunnel egresses the tunnel's own socket. For base and + // vanilla tunnels sockIdx is 0 (stock behavior); for lanes it keeps + // keepalives, close packets and rejects on the lane's 4-tuple so the + // peer's spoof/roam checks accept them and the NAT entry stays warm. + q := hostinfo.sockIdx useRelay := !remote.IsValid() && !hostinfo.GetRemote().IsValid() fullOut := out @@ -565,7 +592,8 @@ func (f *Interface) sendNoMetrics(t header.MessageType, st header.MessageSubType // Query our LH if we haven't since the last time we've been rebound, this will cause the remote to punch against // all our addrs and enable a faster roaming. - if t != header.CloseTunnel && hostinfo.lastRebindCount != f.rebindCount { + // Lanes skip this: the base tunnel issues the one query for the peer. + if t != header.CloseTunnel && !hostinfo.isLane() && hostinfo.lastRebindCount != f.rebindCount { //NOTE: there is an update hole if a tunnel isn't used and exactly 256 rebinds occur before the tunnel is // finally used again. This tunnel would eventually be torn down and recreated if this action didn't help. f.lightHouse.QueryServer(hostinfo.vpnAddrs[0]) @@ -608,6 +636,13 @@ 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 06778587..1783179c 100644 --- a/interface.go +++ b/interface.go @@ -42,10 +42,13 @@ type InterfaceConfig struct { DropLocalBroadcast bool DropMulticast bool routines int - MessageMetrics *MessageMetrics - version string - relayManager *relayManager - punchy *Punchy + // Multiport means writers[i] is bound to listen.port+i (not a shared + // SO_REUSEPORT port) and lane tunnels are negotiated with capable peers. + Multiport bool + MessageMetrics *MessageMetrics + version string + relayManager *relayManager + punchy *Punchy tryPromoteEvery uint32 reQueryEvery uint32 @@ -86,6 +89,7 @@ type Interface struct { dropLocalBroadcast bool dropMulticast bool routines int + multiport bool disconnectInvalid atomic.Bool closed atomic.Bool // cpuAffinity, when non-empty, names the CPUs each TUN reader goroutine @@ -222,6 +226,7 @@ func NewInterface(ctx context.Context, c *InterfaceConfig) (*Interface, error) { dropLocalBroadcast: c.DropLocalBroadcast, dropMulticast: c.DropMulticast, routines: c.routines, + multiport: c.Multiport, version: c.version, writers: make([]udp.Conn, c.routines), batchers: make([]batch.RxBatcher, c.routines), @@ -277,7 +282,10 @@ func (f *Interface) activate() error { "boringcrypto", boringEnabled(), ) - if f.routines > 1 && !f.outside.SupportsMultipleReaders() { + // Under multiport each socket has exactly one reader on its own port, so + // the shared-port multi-reader capability is irrelevant (and main.go + // already hard-errored on unsupported platforms). + if f.routines > 1 && !f.multiport && !f.outside.SupportsMultipleReaders() { f.routines = 1 f.l.Warn("multiple udp readers are not supported on this platform, falling back to a single routine") } @@ -290,6 +298,11 @@ func (f *Interface) activate() error { return err } if len(queues) < f.routines { + if f.multiport { + // The lane sockets are already bound one-per-routine; shrinking + // the routine count would leave bound ports with no reader. + return fmt.Errorf("multiport requires %d tun queues, device provided %d", f.routines, len(queues)) + } f.l.Warn("tun multiqueue is not supported on this platform, falling back to fewer routines", "requested", f.routines, "opened", len(queues)) f.routines = len(queues) @@ -372,7 +385,7 @@ func (f *Interface) listenOut(i int) { scratch := make([]byte, mtu) listener := func(fromUdpAddr netip.AddrPort, payload []byte, meta udp.RxMeta) { - f.readOutsidePackets(ViaSender{UdpAddr: fromUdpAddr}, scratch, payload, h, fwPacket, lhh, nb, i, ctCache.Get(), meta) + f.readOutsidePackets(ViaSender{UdpAddr: fromUdpAddr, SockIdx: i}, scratch, payload, h, fwPacket, lhh, nb, i, ctCache.Get(), meta) } flusher := func() { @@ -394,6 +407,39 @@ func (f *Interface) listenOut(i int) { f.l.Debug("underlay reader is done", "reader", i) } +// txQueue is the per-routine TX state owned by one listenIn goroutine. lane +// is bound to the routine's own 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 on routine 0. +// Concurrent sendmmsg on the shared socket-0 fd is safe: a flow is pinned to +// one routine by tun steering, so per-flow wire order still holds. +type txQueue struct { + lane *batch.SendBatch + base *batch.SendBatch +} + +func (tx *txQueue) full() bool { + if tx.lane.Len() >= batch.SendBatchCap { + return true + } + return tx.base != tx.lane && tx.base.Len() >= batch.SendBatchCap +} + +// 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. +func (tx *txQueue) flush(l *slog.Logger, i int) { + if tx.base != tx.lane { + if err := tx.base.Flush(); err != nil { + l.Error("Failed to write outgoing batch", "error", err, "writer", 0) + } + } + if err := tx.lane.Flush(); err != nil { + l.Error("Failed to write outgoing batch", "error", err, "writer", i) + } +} + func (f *Interface) listenIn(queue tio.Queue, i int) { // Pinning this thread (and goroutine) to a single CPU keeps every sendmmsg from this goroutine going through the // same TX ring on the nic, so the wire sees per-flow order. Skip entirely when tun.pin_threads is false. @@ -419,6 +465,10 @@ func (f *Interface) listenIn(queue tio.Queue, i int) { rejectBuf := make([]byte, mtu) arenaSize := batch.SendBatchCap * (udp.MTU + 32) sb := batch.NewSendBatch(f.writers[i], batch.SendBatchCap, arenaSize) + tx := &txQueue{lane: sb, base: sb} + if f.multiport && i != 0 { + tx.base = batch.NewSendBatch(f.writers[0], batch.SendBatchCap, arenaSize) + } fwPacket := &firewall.Packet{} nb := make([]byte, 12, 12) @@ -436,19 +486,15 @@ func (f *Interface) listenIn(queue tio.Queue, i int) { } for _, pkt := range pkts { - f.consumeInsidePacket(pkt, fwPacket, nb, sb, rejectBuf, i, conntrackCache.Get()) + f.consumeInsidePacket(pkt, fwPacket, nb, tx, rejectBuf, i, conntrackCache.Get()) // Flush incrementally once a full sendmmsg batch has // accumulated so the first packets of a deep read drain // hit the wire while the rest are still being encrypted. - if sb.Len() >= batch.SendBatchCap { - if err := sb.Flush(); err != nil { - f.l.Error("Failed to write outgoing batch", "error", err, "writer", i) - } + if tx.full() { + tx.flush(f.l, i) } } - if err := sb.Flush(); err != nil { - f.l.Error("Failed to write outgoing batch", "error", err, "writer", i) - } + tx.flush(f.l, i) } f.l.Debug("overlay reader is done", "reader", i) diff --git a/lanes_test.go b/lanes_test.go new file mode 100644 index 00000000..9a0f7de0 --- /dev/null +++ b/lanes_test.go @@ -0,0 +1,409 @@ +package nebula + +import ( + "net/netip" + "testing" + "time" + + "github.com/gaissmai/bart" + "github.com/slackhq/nebula/cert" + "github.com/slackhq/nebula/config" + "github.com/slackhq/nebula/header" + "github.com/slackhq/nebula/overlay/batch" + "github.com/slackhq/nebula/overlay/overlaytest" + "github.com/slackhq/nebula/overlay/tio" + "github.com/slackhq/nebula/test" + "github.com/slackhq/nebula/udp" + "github.com/stretchr/testify/assert" + "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) + return base +} + +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), + HandshakePacket: map[uint8][]byte{}, + sockIdx: int(laneIndex), + laneIndex: laneIndex, + laneOwned: owned, + parent: base, + } + lane.SetRemote(netip.MustParseAddrPort("192.0.2.1:4243")) + return lane +} + +func TestLaneHostmapLifecycle(t *testing.T) { + l := test.NewLogger() + hostMap := newHostMap(l) + ifce := &Interface{l: l} // connectionManager nil is tolerated by unlockedAddLane + + vpnIp := netip.MustParseAddr("172.1.1.2") + base := newTestBaseHostInfo(vpnIp, 100, 200, 4) + + hostMap.Lock() + hostMap.unlockedAddHostInfo(base, ifce) + hostMap.Unlock() + + lane := newTestLaneHostInfo(base, 1, 101, 201, true) + hostMap.Lock() + hostMap.unlockedAddLane(lane, ifce) + hostMap.Unlock() + base.lanes.txLanes[1].Store(lane) + + // 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() +} + +func TestLaneHostmapCascadeDelete(t *testing.T) { + l := test.NewLogger() + hostMap := newHostMap(l) + ifce := &Interface{l: l} + + vpnIp := netip.MustParseAddr("172.1.1.3") + base := newTestBaseHostInfo(vpnIp, 300, 400, 4) + + hostMap.Lock() + hostMap.unlockedAddHostInfo(base, ifce) + hostMap.Unlock() + + 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() + + // 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]) +} + +// 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) + + vpnIp := netip.MustParseAddr("172.1.1.4") + pendingBase := hm.StartHandshake(vpnIp, nil) + require.NotNil(t, pendingBase) + + other := &HostInfo{vpnAddrs: []netip.Addr{vpnIp}, localIndexId: 999} + hm.DeleteHostInfo(other) + + // 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)) +} + +func newLaneTestConnectionManager(hostMap *HostMap) (*connectionManager, *Interface) { + l := test.NewLogger() + lh := newTestLighthouse() + cs := &CertState{ + initiatingVersion: cert.Version1, + privateKey: []byte{}, + v1Cert: &dummyCert{version: cert.Version1}, + v1Credential: nil, + } + ifce := &Interface{ + hostMap: hostMap, + inside: &overlaytest.NoopTun{}, + outside: &udp.NoopConn{}, + firewall: &Firewall{}, + lightHouse: lh, + pki: &PKI{}, + handshakeManager: NewHandshakeManager(l, hostMap, lh, &udp.NoopConn{}, defaultHandshakeConfig), + myVpnNetworksTable: new(bart.Lite), + 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) + cm.intf = ifce + ifce.connectionManager = cm + ifce.handshakeManager.f = ifce + return cm, ifce +} + +func TestLaneTrafficDecision(t *testing.T) { + hostMap := newHostMap(test.NewLogger()) + cm, ifce := newLaneTestConnectionManager(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) + + now := time.Now() + + // A lane with inbound traffic is alive and never swaps primary or + // migrates relays. + lane.in.Store(true) + decision, resolved, _ := cm.makeTrafficDecision(lane.localIndexId, now) + assert.Equal(t, doNothing, decision) + assert.Equal(t, lane, resolved) + assert.False(t, lane.pendingDeletion.Load()) + + // An idle lane gets an active keepalive test... + decision, _, _ = cm.makeTrafficDecision(lane.localIndexId, now) + assert.Equal(t, sendTestPacket, decision) + assert.True(t, lane.pendingDeletion.Load()) + + // ...and is declared dead when the test goes unanswered. + decision, _, _ = cm.makeTrafficDecision(lane.localIndexId, now) + assert.Equal(t, deleteTunnel, decision) +} + +func TestBaseInactiveConsidersLanes(t *testing.T) { + hostMap := newHostMap(test.NewLogger()) + cm, _ := newLaneTestConnectionManager(hostMap) + cm.dropInactive.Store(true) + cm.inactivityTimeout.Store(int64(10 * time.Minute)) + + now := time.Now() + vpnIp := netip.MustParseAddr("172.1.1.6") + base := newTestBaseHostInfo(vpnIp, 700, 800, 4) + base.lastUsed = now.Add(-time.Hour) + + // Base alone: inactive. + _, inactive := cm.isInactive(base, now) + assert.True(t, inactive) + + // 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) + + // 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) +} + +// recordingBatchWriter satisfies batch's writer interface and records what +// was flushed to it. +type recordingBatchWriter struct { + bufs [][]byte + dsts []netip.AddrPort +} + +func (w *recordingBatchWriter) WriteBatch(bufs [][]byte, addrs []netip.AddrPort, outerECNs []byte) error { + for i := range bufs { + w.bufs = append(w.bufs, append([]byte(nil), bufs[i]...)) + w.dsts = append(w.dsts, addrs[i]) + } + return nil +} + +func TestSendInsideMessageLaneSwap(t *testing.T) { + hostMap := newHostMap(test.NewLogger()) + cm, ifce := newLaneTestConnectionManager(hostMap) + _ = cm + + 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) + base.ConnectionState = newConnectionStateFromResult(baseInit) + lane.ConnectionState = newConnectionStateFromResult(laneInit) + + baseWriter := &recordingBatchWriter{} + laneWriter := &recordingBatchWriter{} + tx := &txQueue{ + base: batch.NewSendBatch(baseWriter, batch.SendBatchCap, 1<<16), + lane: batch.NewSendBatch(laneWriter, batch.SendBatchCap, 1<<16), + } + + pkt := tio.Packet{Bytes: []byte{0x45, 0, 0, 4, 1, 2, 3, 4}} + nb := make([]byte, 12) + + // With the lane published, routine 1's traffic uses the lane session and + // the lane batch. + base.lanes.txLanes[1].Store(lane) + ifce.sendInsideMessage(base, pkt, nb, tx, 1) + tx.flush(ifce.l, 1) + require.Len(t, laneWriter.bufs, 1) + require.Empty(t, baseWriter.bufs) + assert.Equal(t, lane.GetRemote(), laneWriter.dsts[0]) + + h := &header.H{} + require.NoError(t, h.Parse(laneWriter.bufs[0])) + assert.Equal(t, lane.remoteIndexId, h.RemoteIndex) + + // Routine 2 has no lane: base tunnel, base batch. + ifce.sendInsideMessage(base, pkt, nb, tx, 2) + tx.flush(ifce.l, 1) + 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, tx, 1) + tx.flush(ifce.l, 1) + require.Len(t, baseWriter.bufs, 2) + require.Len(t, laneWriter.bufs, 1) +} + +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() + + hm := ifce.handshakeManager + hm.config.laneCount = 4 + hm.config.lanePortCount = 4 + hm.config.laneBasePort = 4242 + + // A real handshake supplies usable keys and a peer cert. + _, respR := runTestHandshake(t) + respR.PeerLaneIndex = 2 + respR.PeerPortCount = 4 + respR.PeerBasePort = 5353 + + 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")} + + // 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) + + 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) + } +} diff --git a/main.go b/main.go index 5439b773..fbcc41b8 100644 --- a/main.go +++ b/main.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "log/slog" + "math" "net" "net/netip" "runtime/debug" @@ -134,6 +135,22 @@ func Main(c *config.C, configTest bool, buildVersion string, l *slog.Logger, dev udpConns := make([]udp.Conn, routines) port := c.GetInt("listen.port", 0) + // Multiport lanes: bind `routines` consecutive UDP ports (listen.port+i) + // instead of SO_REUSEPORT sharing one, and negotiate one extra tunnel per + // routine with capable peers so each routine's traffic rides its own + // underlay 5-tuple. Defaults on, degrading gracefully when preconditions + // aren't met — managed deployments (dnclient) can't be hard-errored on + // config they don't control. + multiport := c.GetBool("multiport.enabled", true) + if multiport && routines < 2 { + l.Info("multiport disabled: requires routines > 1") + multiport = false + } + if multiport && port != 0 && port+routines-1 > math.MaxUint16 { + l.Warn("multiport disabled: would bind ports beyond 65535", "listen.port", port, "routines", routines) + multiport = false + } + // Callers get no handle to these until the Control is returned, release them on any error. defer func() { if reterr != nil { @@ -163,25 +180,77 @@ func Main(c *config.C, configTest bool, buildVersion string, l *slog.Logger, dev listenHost = ips[0].Unmap() } - for i := 0; i < routines; i++ { - l.Info("listening", "addr", netip.AddrPortFrom(listenHost, uint16(port))) - udpServer, err := udp.NewListener(l, listenHost, port, routines > 1, c.GetInt("listen.batch", 64)) - if err != nil { - return nil, util.NewContextualError("Failed to open udp listener", m{"queue": i}, err) - } - udpServer.ReloadConfig(c) - udpConns[i] = udpServer - - // If port is dynamic, discover it before the next pass through the for loop - // This way all routines will use the same port correctly - if port == 0 { - uPort, err := udpServer.LocalAddr() - if err != nil { - return nil, util.NewContextualError("Failed to get listening port", nil, err) + // Every lane socket needs its own reader; a platform that can't run + // multiple readers would silently strand sockets 1..n-1 as blackholes. + // Probe capability before committing to per-port binds. + if multiport { + probe, err := udp.NewListener(l, listenHost, 0, false, 1) + if err == nil { + if !probe.SupportsMultipleReaders() { + l.Warn("multiport disabled: this platform does not support multiple udp readers") + multiport = false } - port = int(uPort.Port()) + _ = probe.Close() } } + + // With a dynamic listen.port, multiport binds socket 0 dynamically and + // then claims the next routines-1 ports above it; if that range turns + // out to be partially occupied, re-roll with a fresh dynamic port. + dynamic := port == 0 + var bindErr error + for attempt := 0; attempt < 6; attempt++ { + bindErr = nil + for i := 0; i < routines; i++ { + lPort := port + if multiport { + lPort = port + i + } + udpServer, err := udp.NewListener(l, listenHost, lPort, routines > 1 && !multiport, c.GetInt("listen.batch", 64)) + if err != nil { + bindErr = util.NewContextualError("Failed to open udp listener", m{"queue": i}, err) + break + } + udpServer.ReloadConfig(c) + udpConns[i] = udpServer + + // If port is dynamic, discover it before the next pass through the for loop + // This way all routines will use the same port correctly + if port == 0 { + uPort, err := udpServer.LocalAddr() + if err != nil { + return nil, util.NewContextualError("Failed to get listening port", nil, err) + } + port = int(uPort.Port()) + if multiport && port+routines-1 > math.MaxUint16 { + bindErr = util.NewContextualError("multiport dynamic port too close to 65535", m{"port": port}, nil) + break + } + } + bound := port + if multiport { + bound = port + i + } + l.Info("listening", "addr", netip.AddrPortFrom(listenHost, uint16(bound)), "socket", i) + } + if bindErr == nil { + break + } + if !(multiport && dynamic) { + return nil, bindErr + } + for i := range udpConns { + if udpConns[i] != nil { + _ = udpConns[i].Close() + udpConns[i] = nil + } + } + port = 0 + l.Debug("multiport dynamic port range collided, retrying", "attempt", attempt+1) + } + if bindErr != nil { + return nil, bindErr + } } hostMap := NewHostMapFromConfig(l, c) @@ -206,6 +275,16 @@ func Main(c *config.C, configTest bool, buildVersion string, l *slog.Logger, dev messageMetrics: messageMetrics, } + if multiport { + lanes := c.GetInt("multiport.lanes", 0) + if lanes <= 0 || lanes > routines { + lanes = routines + } + handshakeConfig.laneCount = lanes + handshakeConfig.lanePortCount = uint16(routines) + handshakeConfig.laneBasePort = uint16(port) + } + handshakeManager := NewHandshakeManager(l, hostMap, lightHouse, udpConns[0], handshakeConfig) lightHouse.handshakeTrigger = handshakeManager.trigger @@ -230,6 +309,7 @@ func Main(c *config.C, configTest bool, buildVersion string, l *slog.Logger, dev DropLocalBroadcast: c.GetBool("tun.drop_local_broadcast", false), DropMulticast: c.GetBool("tun.drop_multicast", false), routines: routines, + Multiport: multiport, MessageMetrics: messageMetrics, version: buildVersion, relayManager: NewRelayManager(ctx, l, hostMap, c), diff --git a/outside.go b/outside.go index 4f37a766..5acf8e0d 100644 --- a/outside.go +++ b/outside.go @@ -101,7 +101,7 @@ func (f *Interface) readOutsidePackets(via ViaSender, scratch []byte, packet []b // recvError if necessary if hostinfo == nil || hostinfo.ConnectionState == nil { if !via.IsRelayed { - f.maybeSendRecvError(via.UdpAddr, h.RemoteIndex) + f.maybeSendRecvError(via.UdpAddr, h.RemoteIndex, q) } return } @@ -203,6 +203,7 @@ func (f *Interface) handleOutsideRelayPacket(hostinfo *HostInfo, via ViaSender, relayHI: hostinfo, relay: relay, IsRelayed: true, + SockIdx: via.SockIdx, } f.readOutsidePackets(via, scratch, signedPayload, h, fwPacket, lhf, nb, q, localCache, meta) case ForwardingType: @@ -595,7 +596,7 @@ func (f *Interface) handleOutsideMessagePacket(hostinfo *HostInfo, out []byte, s dropReason := f.firewall.Drop(*fwPacket, true, hostinfo, f.pki.GetCAPool(), localCache) if dropReason != nil { - f.rejectOutside(out, hostinfo.ConnectionState, hostinfo, nb, scratch, q) + f.rejectOutside(out, hostinfo.ConnectionState, hostinfo, nb, scratch) if f.l.Enabled(context.Background(), slog.LevelDebug) { hostinfo.logger(f.l).Debug("dropping inbound packet", "fwPacket", fwPacket, @@ -611,17 +612,20 @@ func (f *Interface) handleOutsideMessagePacket(hostinfo *HostInfo, out []byte, s } } -func (f *Interface) maybeSendRecvError(endpoint netip.AddrPort, index uint32) { +func (f *Interface) maybeSendRecvError(endpoint netip.AddrPort, index uint32, q int) { if f.sendRecvErrorConfig.ShouldRecvError(endpoint) { - f.sendRecvError(endpoint, index) + f.sendRecvError(endpoint, index, q) } } -func (f *Interface) sendRecvError(endpoint netip.AddrPort, index uint32) { +// sendRecvError replies from the socket the offending packet arrived on (q). +// A lane peer's spoof guard compares our source addr against the lane's +// remote, so a reply from the base port would be discarded. +func (f *Interface) sendRecvError(endpoint netip.AddrPort, index uint32, q int) { f.messageMetrics.Tx(header.RecvError, 0, 1) b := header.Encode(make([]byte, header.Len), header.Version, header.RecvError, 0, index, 0) - _ = f.outside.WriteTo(b, endpoint) + _ = f.writers[q].WriteTo(b, endpoint) if f.l.Enabled(context.Background(), slog.LevelDebug) { f.l.Debug("Recv error sent", "index", index,