the definitive tun offloads branch

Makefile

@@ -161,6 +161,10 @@ bin-pkcs11: BUILD_ARGS += -tags pkcs11
 bin-pkcs11: CGO_ENABLED = 1
 bin-pkcs11: bin
 
+# Build with the pprof debug server (serves on :6060). See startPprofServer.
+debug: BUILD_ARGS += -tags debug
+debug: bin
+
 bin:
 	go build $(BUILD_ARGS) -ldflags "$(LDFLAGS)" -o ./nebula${NEBULA_CMD_SUFFIX} ${NEBULA_CMD_PATH}
 	go build $(BUILD_ARGS) -ldflags "$(LDFLAGS)" -o ./nebula-cert${NEBULA_CMD_SUFFIX} ./cmd/nebula-cert
@@ -280,5 +284,5 @@ smoke-vagrant/%: bin-docker build/%/nebula
 	cd .github/workflows/smoke/ && ./smoke-vagrant.sh $*
 
 .FORCE:
-.PHONY: all all-linux all-freebsd all-openbsd all-netbsd all-darwin all-windows all-cross-linux all-cross-linux-arm all-cross-linux-mips all-cross-linux-other all-cross-darwin all-cross-windows bench bench-cpu bench-cpu-long bin build-test-mobile e2e e2ev e2evv e2evvv e2evvvv proto release service smoke-docker smoke-docker-race test test-cov-html smoke-vagrant/%
+.PHONY: all all-linux all-freebsd all-openbsd all-netbsd all-darwin all-windows all-cross-linux all-cross-linux-arm all-cross-linux-mips all-cross-linux-other all-cross-darwin all-cross-windows bench bench-cpu bench-cpu-long bin debug build-test-mobile e2e e2ev e2evv e2evvv e2evvvv proto release service smoke-docker smoke-docker-race test test-cov-html smoke-vagrant/%
 .DEFAULT_GOAL := bin

connection_state.go

@@ -12,7 +12,17 @@ import (
 	"github.com/slackhq/nebula/noiseutil"
 )
 
-const ReplayWindow = 1024
+const ReplayWindow = 8192
+
+// sessionEpoch hands out a receiver-local ordinal to every ConnectionState as
+// it is created. The RX staging sort (overlay/batch) orders packets by
+// (epoch, message counter). A re-handshake never rekeys an existing tunnel; it
+// brings up a whole new hostinfo (and ConnectionState) with its own counter
+// space starting near zero, while the old tunnel keeps decrypting in-flight
+// packets until it is torn down. During that cutover one flush batch can hold
+// packets from both tunnels, and the epoch is what keeps the old tunnel's
+// packets sorted ahead of the new tunnel's.
+var sessionEpoch atomic.Uint64
 
 type ConnectionState struct {
 	eKey           noiseutil.CipherState
@@ -24,6 +34,8 @@ type ConnectionState struct {
 	window         *Bits
 	decryptLock    sync.Mutex
 	writeLock      sync.Mutex
+	// epoch is this session's sessionEpoch ordinal. Immutable after creation.
+	epoch uint64
 }
 
 // newConnectionStateFromResult builds a fully-populated ConnectionState from a
@@ -38,6 +50,7 @@ func newConnectionStateFromResult(r *handshake.Result) *ConnectionState {
 		eKey:      noiseutil.NewCipherState(r.EKey, r.Cipher),
 		dKey:      noiseutil.NewCipherState(r.DKey, r.Cipher),
 		window:    NewBits(ReplayWindow),
+		epoch:     sessionEpoch.Add(1),
 	}
 	ci.messageCounter.Add(r.MessageIndex)
 	for i := uint64(1); i <= r.MessageIndex; i++ {
@@ -58,16 +71,15 @@ func (cs *ConnectionState) Curve() cert.Curve {
 	return cs.myCert.Curve()
 }
 
-func (cs *ConnectionState) Decrypt(l *slog.Logger, messageCounter uint64, out []byte, packet []byte, nb []byte) ([]byte, error) {
-	var err error
+func (cs *ConnectionState) Decrypt(l *slog.Logger, messageCounter uint64, packet []byte, nb []byte) ([]byte, error) {
 	cs.decryptLock.Lock()
 	result := cs.window.Check(l, messageCounter)
 	cs.decryptLock.Unlock()
 	if !result {
 		return nil, ErrAlreadySeen
 	}
 
-	out, err = cs.dKey.DecryptDanger(out, packet[:header.Len], packet[header.Len:], messageCounter, nb)
+	out, err := cs.dKey.DecryptDanger(packet[header.Len:header.Len], packet[:header.Len], packet[header.Len:], messageCounter, nb)
 	if err != nil {
 		return nil, err
 	}
@@ -81,7 +93,6 @@ func (cs *ConnectionState) Decrypt(l *slog.Logger, messageCounter uint64, out []
 	return out, nil
 }
 
-// VerifyRelay verifies AEAD protected (but not encrypted) relay frames. packet must be length-checked by the caller.
 func (cs *ConnectionState) VerifyRelay(l *slog.Logger, messageCounter uint64, packet []byte, nb []byte) error {
 	cs.decryptLock.Lock()
 	result := cs.window.Check(l, messageCounter)
@@ -90,6 +101,11 @@ func (cs *ConnectionState) VerifyRelay(l *slog.Logger, messageCounter uint64, pa
 		return ErrAlreadySeen
 	}
 
+	// The entire body is sent as AD, not encrypted.
+	// The packet consists of a 16-byte parsed Nebula header, Associated Data-protected payload, and a trailing 16-byte AEAD signature value.
+	// The packet is guaranteed to be at least 16 bytes at this point, b/c it got past the h.Parse() call above. If it's
+	// otherwise malformed (meaning, there is no trailing 16 byte AEAD value), then this will result in at worst a 0-length slice
+	// which will gracefully fail in the DecryptDanger call.
 	signedPayload := packet[:len(packet)-cs.dKey.Overhead()]
 	signatureValue := packet[len(packet)-cs.dKey.Overhead():]
 	_, err := cs.dKey.DecryptDanger(nil, signedPayload, signatureValue, messageCounter, nb)
@@ -103,6 +119,5 @@ func (cs *ConnectionState) VerifyRelay(l *slog.Logger, messageCounter uint64, pa
 	if !result {
 		return ErrAlreadySeen
 	}
-
 	return nil
 }

control_lifecycle_test.go

@@ -11,6 +11,8 @@ import (
 
 	"github.com/gaissmai/bart"
 	"github.com/slackhq/nebula/config"
+	"github.com/slackhq/nebula/overlay/batch"
+	"github.com/slackhq/nebula/overlay/tio"
 	"github.com/slackhq/nebula/routing"
 	"github.com/slackhq/nebula/test"
 	"github.com/slackhq/nebula/udp"
@@ -30,9 +32,9 @@ func newFakeDevice() *fakeDevice {
 
 // Read blocks until Close like a real tun with no traffic, then reports EOF
 // the same way a closed device does
-func (d *fakeDevice) Read(p []byte) (int, error) {
+func (d *fakeDevice) Read() ([]tio.Packet, error) {
 	<-d.closedCh
-	return 0, io.EOF
+	return nil, io.EOF
 }
 
 func (d *fakeDevice) Write(p []byte) (int, error) { return len(p), nil }
@@ -49,10 +51,8 @@ func (d *fakeDevice) Activate() error                       { return nil }
 func (d *fakeDevice) Networks() []netip.Prefix              { return nil }
 func (d *fakeDevice) Name() string                          { return "fake" }
 func (d *fakeDevice) RoutesFor(netip.Addr) routing.Gateways { return nil }
-func (d *fakeDevice) SupportsMultiqueue() bool              { return false }
-func (d *fakeDevice) NewMultiQueueReader() (io.ReadWriteCloser, error) {
-	return nil, errors.New("unsupported")
-}
+
+func (d *fakeDevice) Queues(int) ([]tio.Queue, error) { return []tio.Queue{d}, nil }
 
 // newReadyControl hand-builds the minimum Control that Main would have
 // produced right before Start, including the construction token NewInterface
@@ -78,7 +78,7 @@ func newReadyControl(t *testing.T) (*Control, *fakeDevice, *fakeConn) {
 		inside:     dev,
 		outside:    conn,
 		writers:    []udp.Conn{conn},
-		readers:    make([]io.ReadWriteCloser, 1),
+		batchers:   make([]batch.RxBatcher, 1),
 		routines:   1,
 		hostMap:    newHostMap(l),
 		lightHouse: lh,
@@ -109,7 +109,8 @@ func TestControl_StopBeforeStart(t *testing.T) {
 	require.NoError(t, c.Wait())
 
 	// A stopped control can never be started
-	require.ErrorIs(t, c.Start(), ErrAlreadyStopped)
+	err := c.Start()
+	require.ErrorIs(t, err, ErrAlreadyStopped)
 
 	// A second Stop is a harmless no-op
 	c.Stop()
@@ -143,19 +144,29 @@ type fakeConn struct {
 	rebinds int
 }
 
-func (c *fakeConn) Rebind() error                            { c.rebinds++; return nil }
-func (c *fakeConn) LocalAddr() (netip.AddrPort, error)       { return netip.AddrPort{}, nil }
-func (c *fakeConn) ListenOut(_ udp.EncReader) error          { return nil }
-func (c *fakeConn) WriteTo(_ []byte, _ netip.AddrPort) error { return nil }
-func (c *fakeConn) ReloadConfig(_ *config.C)                 {}
-func (c *fakeConn) SupportsMultipleReaders() bool            { return true }
-func (c *fakeConn) Close() error                             { c.closed = true; return nil }
+func (c *fakeConn) Rebind() error                             { c.rebinds++; return nil }
+func (c *fakeConn) LocalAddr() (netip.AddrPort, error)        { return netip.AddrPort{}, nil }
+func (c *fakeConn) ListenOut(_ udp.EncReader, _ func()) error { return nil }
+func (c *fakeConn) WriteTo(_ []byte, _ netip.AddrPort) error  { return nil }
+func (c *fakeConn) WriteBatch(bufs [][]byte, _ []netip.AddrPort) (int, error) {
+	return len(bufs), nil
+}
+func (c *fakeConn) ReloadConfig(_ *config.C)      {}
+func (c *fakeConn) SupportsMultipleReaders() bool { return true }
+func (c *fakeConn) Close() error                  { c.closed = true; return nil }
 
 type multiqueueDevice struct {
 	*fakeDevice
 }
 
-func (d *multiqueueDevice) SupportsMultiqueue() bool { return true }
+// Queues claims multiqueue support but fails to open the second queue,
+// exercising the activation error path.
+func (d *multiqueueDevice) Queues(n int) ([]tio.Queue, error) {
+	if n > 1 {
+		return nil, errors.New("second queue failed to open")
+	}
+	return d.fakeDevice.Queues(n)
+}
 
 func TestControl_StartMultiqueueFailureReleases(t *testing.T) {
 	dev := &multiqueueDevice{fakeDevice: newFakeDevice()}
@@ -166,7 +177,7 @@ func TestControl_StartMultiqueueFailureReleases(t *testing.T) {
 		inside:   dev,
 		outside:  conn,
 		writers:  []udp.Conn{conn},
-		readers:  make([]io.ReadWriteCloser, 2),
+		batchers: make([]batch.RxBatcher, 2),
 		routines: 2,
 		l:        test.NewLogger(),
 	}
@@ -181,7 +192,8 @@ func TestControl_StartMultiqueueFailureReleases(t *testing.T) {
 	}
 
 	// The second reader fails to open, everything must be released
-	require.Error(t, c.Start())
+	err := c.Start()
+	require.Error(t, err)
 	assert.Equal(t, StateStopped, c.State())
 	assert.True(t, dev.closed, "the tun device should have been closed")
 	assert.True(t, conn.closed, "the udp socket should have been closed")
@@ -251,15 +263,18 @@ func TestControl_ConcurrentStopAndStart(t *testing.T) {
 	// panic and Wait must observe the final state
 	require.NoError(t, c.Wait())
 	assert.Equal(t, StateStopped, c.State())
-	require.ErrorIs(t, c.Start(), ErrAlreadyStopped)
+	err := c.Start()
+	require.ErrorIs(t, err, ErrAlreadyStopped)
 }
 
 func TestControl_StartStopLifecycle(t *testing.T) {
 	c, dev, conn := newReadyControl(t)
 
-	require.NoError(t, c.Start())
+	err := c.Start()
+	require.NoError(t, err)
 	assert.Equal(t, StateStarted, c.State())
-	require.ErrorIs(t, c.Start(), ErrAlreadyStarted)
+	err = c.Start()
+	require.ErrorIs(t, err, ErrAlreadyStarted)
 
 	// Stop must unpark the reader blocked in the device and release everything
 	c.Stop()
@@ -270,7 +285,8 @@ func TestControl_StartStopLifecycle(t *testing.T) {
 
 	// The reader drained off a closed device, that is not a fatal error
 	require.NoError(t, c.Wait())
-	require.ErrorIs(t, c.Start(), ErrAlreadyStopped)
+	err = c.Start()
+	require.ErrorIs(t, err, ErrAlreadyStopped)
 }
 
 func TestControl_RebindIsGatedByState(t *testing.T) {
@@ -280,7 +296,8 @@ func TestControl_RebindIsGatedByState(t *testing.T) {
 	c.RebindUDPServer()
 	assert.Equal(t, 0, conn.rebinds, "rebind before start must be a no-op")
 
-	require.NoError(t, c.Start())
+	err := c.Start()
+	require.NoError(t, err)
 	c.RebindUDPServer()
 	assert.Equal(t, 1, conn.rebinds, "rebind while started must reach the conn")