diff --git a/connection_state.go b/connection_state.go index 3018048b..b949fc1d 100644 --- a/connection_state.go +++ b/connection_state.go @@ -104,7 +104,7 @@ func (cs *ConnectionState) Decrypt(l *slog.Logger, messageCounter uint64, packet return nil, ErrAlreadySeen } - out, err := cs.dKey.DecryptDanger(packet[header.Len:header.Len], packet[:header.Len], packet[header.Len:], messageCounter, nb) + out, err := cs.decryptPacket(messageCounter, packet, nb) if err != nil { return nil, err } @@ -118,6 +118,11 @@ func (cs *ConnectionState) Decrypt(l *slog.Logger, messageCounter uint64, packet return out, nil } +// decryptPacket authenticates and decrypts packet in place, leaving the replay window to the caller. +func (cs *ConnectionState) decryptPacket(messageCounter uint64, packet, nb []byte) ([]byte, error) { + return cs.dKey.DecryptDanger(packet[header.Len:header.Len], packet[:header.Len], packet[header.Len:], messageCounter, nb) +} + func (cs *ConnectionState) VerifyRelay(l *slog.Logger, messageCounter uint64, packet []byte, nb []byte) error { cs.decryptLock.Lock() result := cs.window.Check(l, messageCounter) diff --git a/connection_state_test.go b/connection_state_test.go index a946343d..1f34a206 100644 --- a/connection_state_test.go +++ b/connection_state_test.go @@ -23,11 +23,20 @@ import ( // runTestHandshake runs a complete IX handshake between two freshly-built // peers and returns the initiator and responder Results. Used to produce // real cipher states for tests that need to exercise post-handshake glue. -func runTestHandshake(t *testing.T) (initR, respR *handshake.Result) { +func runTestHandshake(t testing.TB) (initR, respR *handshake.Result) { + t.Helper() + return runTestHandshakeWith(t, noise.CipherChaChaPoly) +} + +// runTestHandshakeWith is runTestHandshake with the data plane cipher picked by the caller. +func runTestHandshakeWith(t testing.TB, cipher noise.CipherFunc) (initR, respR *handshake.Result) { t.Helper() + // Valid around now rather than the cert_test default, which is fixed at process start and lapses a minute + // later, inside a long -count run + now := time.Now() ca, _, caKey, _ := ct.NewTestCaCert( - cert.Version2, cert.Curve_CURVE25519, time.Time{}, time.Time{}, nil, nil, nil, + cert.Version2, cert.Curve_CURVE25519, now.Add(-time.Minute), now.Add(time.Hour), nil, nil, nil, ) caPool := ct.NewTestCAPool(ca) @@ -40,7 +49,7 @@ func runTestHandshake(t *testing.T) (initR, respR *handshake.Result) { require.NoError(t, err) hsBytes, err := c.MarshalForHandshakes() require.NoError(t, err) - ncs := noise.NewCipherSuite(noise.DH25519, noise.CipherChaChaPoly, noise.HashSHA256) + ncs := noise.NewCipherSuite(noise.DH25519, cipher, noise.HashSHA256) cred := handshake.NewCredential(c, hsBytes, priv, ncs) return func(v cert.Version) *handshake.Credential { if v == cert.Version2 { diff --git a/e2e/tunnels_test.go b/e2e/tunnels_test.go index 7831b7dd..5b04f13a 100644 --- a/e2e/tunnels_test.go +++ b/e2e/tunnels_test.go @@ -5,9 +5,11 @@ package e2e import ( "fmt" "net/netip" + "runtime" "testing" "time" + "github.com/rcrowley/go-metrics" "github.com/slackhq/nebula/cert" "github.com/slackhq/nebula/cert_test" "github.com/slackhq/nebula/e2e/router" @@ -570,3 +572,29 @@ func TestCloseTunnelAuthenticated(t *testing.T) { myControl.Stop() theirControl.Stop() } + +// TestStagedTunnel completes a handshake and traffic both ways with staged receive. Every received packet, +// handshakes included, uses the receive pipeline on Linux. Other platforms must do the same over the inline fallback. +func TestStagedTunnel(t *testing.T) { + t.Parallel() + received := metrics.GetOrRegisterCounter("flow.rx.inline", nil) + before := received.Count() + staged := m{"tun": m{"staged": m{"workers": 2}}} + ca, _, caKey, _ := cert_test.NewTestCaCert(cert.Version1, cert.Curve_CURVE25519, time.Now(), time.Now().Add(10*time.Minute), nil, nil, []string{}) + myControl, myVpnIpNet, myUdpAddr, _ := newSimpleServer(cert.Version1, ca, caKey, "me", "10.128.0.1/24", staged) + theirControl, theirVpnIpNet, theirUdpAddr, _ := newSimpleServer(cert.Version1, ca, caKey, "them", "10.128.0.2/24", staged) + myControl.InjectLightHouseAddr(theirVpnIpNet[0].Addr(), theirUdpAddr) + theirControl.InjectLightHouseAddr(myVpnIpNet[0].Addr(), myUdpAddr) + myControl.Start() + theirControl.Start() + r := router.NewR(t, myControl, theirControl) + defer r.RenderFlow() + assertTunnel(t, myVpnIpNet[0].Addr(), theirVpnIpNet[0].Addr(), myControl, theirControl, r) + myControl.Stop() + theirControl.Stop() + if runtime.GOOS != "linux" { + assert.Zero(t, received.Count()-before) + return + } + assert.GreaterOrEqual(t, received.Count()-before, int64(2)) +} diff --git a/examples/config.yml b/examples/config.yml index 0d1a9180..b36d60be 100644 --- a/examples/config.yml +++ b/examples/config.yml @@ -271,6 +271,14 @@ tun: # the use_offloads setting controls if Nebula will attempt to enable GSO and GRO for the tun device. Linux only, not reloadable. #use_offloads: false + # Linux only. staged.workers enables staged receive: each routine hands its received batches to this many decrypt + # workers, then an ordered writer commits them in arrival order through the same replay, roaming and firewall + # handling as inline receive. 0, the default, keeps receive inline. Batches below 256 packets on an idle listener + # stay on the listener. Needs no support from peers. Costs about 24 MiB of buffers per routine at listen.batch 64 + # with GRO. Not reloadable. + #staged: + # workers: 0 + # Linux only. pin_threads pins each tun reader/encrypt OS thread to a single CPU. This keeps every goroutine's # batched sends flowing through one XPS-selected NIC TX ring, so packets within a flow stay ordered on the wire # instead of being sprayed across multiple TX rings and reordered. Not reloadable. Coerced to false if routines <= 1. diff --git a/flow.go b/flow.go new file mode 100644 index 00000000..aed76049 --- /dev/null +++ b/flow.go @@ -0,0 +1,80 @@ +package nebula + +import ( + "fmt" + "log/slog" + "runtime" + "sync" + "time" + + "github.com/rcrowley/go-metrics" + "github.com/slackhq/nebula/config" +) + +// Staged receive (tun.staged.workers) moves decryption off each udp listener onto worker goroutines, behind which an +// ordered writer commits every batch in arrival order. It reuses the inline packet handlers: staging changes ownership +// and scheduling, not policy. Each routine has its own workers and writer, so order is per routine. Full queues block +// the listener. A read error drains the batches in flight, while shutdown waits for the workers and then drops what +// is left. Each routine keeps six batches; at listen.batch=64 with GRO their buffers take about 24 MiB per routine. +// +// Metrics are registered only for active pipelines. flow.rx.batches counts batches handed to the writer and +// flow.rx.inline those an idle listener handled itself; their rates show whether staging is used. +// flow.stage..q.busy_ns locates the limiting stage: rx_read, rx_decrypt or rx_write. Take rates; 1e9 +// ns/s fills a serial stage. These are elapsed times that exclude read syscalls and downstream waits; descheduling +// inflates them and worker times add up across the pool. Small batches sent to the writer include decryption in +// writer time. + +// stagedConfig is tun.staged. workers is per routine; zero keeps receive inline. +type stagedConfig struct { + workers int + // inlineBelow is the packet count below which an idle listener skips the worker handoff + inlineBelow int +} + +// parseStagedConfig enables staging on Linux only; conns that cannot hand off batches also stay inline. +func parseStagedConfig(c *config.C, l *slog.Logger) stagedConfig { + sc := stagedConfig{workers: c.GetInt("tun.staged.workers", 0)} + if sc.workers <= 0 { + return stagedConfig{} + } + if runtime.GOOS != "linux" { + l.Warn("tun.staged is only supported on Linux; packet processing stays inline") + return stagedConfig{} + } + // 256 is the measured crossover + sc.inlineBelow = 256 + l.Info("Staged receive processing enabled", "workers", sc.workers) + return sc +} + +// flowJob is one batch of crypto work. It owns its packet buffers until run returns; jobs and queues are reused. +type flowJob interface{ run(nb []byte) } + +// flowPool runs one listener's crypto jobs. The listener is the only producer and closes the pool once its ordered +// stage has drained. +type flowPool struct { + jobs chan flowJob + wg sync.WaitGroup +} + +func newFlowPool(workers, queueLen int, busy metrics.Counter) *flowPool { + p := &flowPool{jobs: make(chan flowJob, queueLen)} + for range workers { + p.wg.Go(func() { + nb := make([]byte, 12) + for job := range p.jobs { + t := time.Now() + job.run(nb) + busy.Inc(int64(time.Since(t))) + } + }) + } + return p +} + +func (p *flowPool) close() { close(p.jobs); p.wg.Wait() } + +// stageBusy is the busy time counter of one stage of routine q. +func stageBusy(stage string, q int) metrics.Counter { + return metrics.GetOrRegisterCounter(fmt.Sprintf("flow.stage.%s.q%d.busy_ns", stage, q), nil) +} diff --git a/flow_rx.go b/flow_rx.go new file mode 100644 index 00000000..2c98b8f1 --- /dev/null +++ b/flow_rx.go @@ -0,0 +1,398 @@ +package nebula + +import ( + "encoding/binary" + "net/netip" + "time" + + "github.com/rcrowley/go-metrics" + "github.com/slackhq/nebula/header" + "github.com/slackhq/nebula/udp" +) + +// handoffConn is a udp conn that can hand each received batch over. The callback returns replacement buffers with +// the same count and at least the same lengths. +type handoffConn interface { + ListenOutHandoff(udp.EncReader, func([][]byte) [][]byte) error +} + +const stagedRxSlots = 6 +const rxChunkMin = 16 + +// rxItem is one received packet. A data packet for an established tunnel has ci set; out and err are written by +// whoever decrypts it. +type rxItem struct { + packet []byte + hostinfo *HostInfo + ci *ConnectionState + counter uint64 + via netip.AddrPort + out []byte + err error +} + +// rxChunk is a contiguous run of a batch's items for one worker to decrypt. +type rxChunk struct { + items []rxItem + done chan struct{} +} + +func (c *rxChunk) run(nb []byte) { + decryptRxItems(c.items, nb) + c.done <- struct{}{} +} + +// decryptRxItems decrypts the data packets among items in place. The replay window is left to the ordered stage, as +// are packets without a tunnel and the known replays precheck marked. +func decryptRxItems(items []rxItem, nb []byte) { + for i := range items { + it := &items[i] + if it.ci == nil || it.err != nil { + continue + } + it.out, it.err = it.ci.decryptPacket(it.counter, it.packet, nb) + } +} + +// rxBatch is one recvmmsg batch moving through the staged receive pipeline. +type rxBatch struct { + // items is every packet of the batch in arrival order. Data packets are decrypted by a worker; anything else is + // handled whole by the writer, in its place. + items []rxItem + data int + // bufs is the set of recvmmsg buffers the items point into, owned by the batch from handoff until it is reused. + bufs [][]byte + // chunks are the decrypt jobs, nChunks of them in use. With none the writer decrypts the batch itself. + chunks []rxChunk + nChunks int +} + +func (b *rxBatch) reset() { + clear(b.items) + b.items = b.items[:0] + b.data = 0 + for i := range b.nChunks { + b.chunks[i].items = nil + } + b.nChunks = 0 +} + +// stagedRx is the staged receive pipeline of one udp listener. deliver and handoff run on the listener goroutine. +type stagedRx struct { + f *Interface + q int + pool *flowPool + // h and hostmapCache are the listener's own. wrxc is the commit context: the writer's, or the listener's while + // nothing is in flight. + h *header.H + hostmapCache map[uint32]*HostInfo + wrxc *rxContext + + started time.Time + cur *rxBatch + free []*rxBatch + ordered chan *rxBatch + returned chan *rxBatch + inFlight int + writerDone chan struct{} + + readBusy, decryptBusy, writeBusy metrics.Counter + metricBatches metrics.Counter + metricInline metrics.Counter +} + +func newStagedRx(f *Interface, q int, wrxc *rxContext) *stagedRx { + p := &stagedRx{ + f: f, + q: q, + h: &header.H{}, + hostmapCache: map[uint32]*HostInfo{}, + wrxc: wrxc, + ordered: make(chan *rxBatch, stagedRxSlots), + returned: make(chan *rxBatch, stagedRxSlots), + writerDone: make(chan struct{}), + readBusy: stageBusy("rx_read", q), + decryptBusy: stageBusy("rx_decrypt", q), + writeBusy: stageBusy("rx_write", q), + metricBatches: metrics.GetOrRegisterCounter("flow.rx.batches", nil), + metricInline: metrics.GetOrRegisterCounter("flow.rx.inline", nil), + } + workers := f.staged.workers + p.pool = newFlowPool(workers, stagedRxSlots*2*workers, p.decryptBusy) + for range stagedRxSlots { + // Two chunks per worker, so the writer can start committing while the workers are still on the rest + b := &rxBatch{items: make([]rxItem, 0, 256), chunks: make([]rxChunk, 2*workers)} + for i := range b.chunks { + b.chunks[i].done = make(chan struct{}, 1) + } + p.free = append(p.free, b) + } + p.cur = p.free[len(p.free)-1] + p.free = p.free[:len(p.free)-1] + go p.writer() + return p +} + +// listen runs the conn's receive loop through the pipeline until it fails, then drains the pipeline. The conn hands +// its buffers over, so a batch moves between stages without a copy. +func (p *stagedRx) listen(li handoffConn) error { + defer p.close() + return li.ListenOutHandoff(p.deliver, p.handoff) +} + +// deliver is the listener's per-packet work: classify the packet and add it to the current batch. Nothing here +// decrypts or touches the tun. +func (p *stagedRx) deliver(via netip.AddrPort, packet []byte) { + if p.started.IsZero() { + p.started = time.Now() + } + b := p.cur + it := rxItem{packet: packet, via: via} + p.classify(&it) + if it.ci != nil { + b.data++ + } + b.items = append(b.items, it) +} + +// classify fills in the tunnel for a plain data packet of an established, non-relayed tunnel, leaving anything else +// for the writer to handle whole. The replay pre-check waits for the end of the batch, see precheck. +func (p *stagedRx) classify(it *rxItem) { + f := p.f + h := p.h + packet := it.packet + if h.Parse(packet) != nil || h.Version != header.Version || h.Type != header.Message || h.Subtype != header.MessageNone { + return + } + if f.myVpnNetworksTable.Contains(it.via.Addr()) { + return + } + hostinfo := f.hostMap.QueryIndexCached(h.RemoteIndex, p.hostmapCache) + if hostinfo == nil || hostinfo.ConnectionState == nil { + return + } + ci := hostinfo.ConnectionState + if len(packet) < header.Len+ci.dKey.Overhead() { + return + } + it.hostinfo, it.ci, it.counter = hostinfo, ci, h.MessageCounter +} + +// precheck marks obvious replays before anything decrypts them, like ConnectionState.Decrypt does, taking each +// tunnel's lock once per run of its packets rather than once per packet: the writer takes the same lock for its +// window updates, from another core. The writer checks again when it commits. The window only moves forward in +// between, so a stale check here can only let a replay through to that second check. Marked packets keep their +// place: an earlier RecvError may delete their tunnel, and inline receive then replies with a RecvError rather than +// dropping a replay. +func (p *stagedRx) precheck(b *rxBatch) { + f := p.f + items := b.items + for i := 0; i < len(items); { + ci := items[i].ci + if ci == nil { + i++ + continue + } + ci.decryptLock.Lock() + for ; i < len(items) && items[i].ci == ci; i++ { + if ci.window.Check(f.l, items[i].counter) { + continue + } + b.data-- + items[i].err = ErrAlreadySeen + } + ci.decryptLock.Unlock() + } +} + +// handoff ends a batch: the batch takes bufs, and the conn reads the next batch into the buffers of the next free +// batch. +func (p *stagedRx) handoff(bufs [][]byte) [][]byte { + p.cur.bufs = bufs + p.endBatch() + if !p.started.IsZero() { + p.readBusy.Inc(int64(time.Since(p.started))) + p.started = time.Time{} + } + b := p.cur + if len(b.bufs) != len(bufs) || len(b.bufs[0]) < len(bufs[0]) { + b.bufs = make([][]byte, len(bufs)) + for i := range b.bufs { + b.bufs[i] = make([]byte, len(bufs[0])) + } + } + return b.bufs +} + +// endBatch hands the batch to the workers and the ordered writer, then takes a free batch. +func (p *stagedRx) endBatch() { + clear(p.hostmapCache) + p.reap() + b := p.cur + p.precheck(b) + n := len(b.items) + if n == 0 { + b.reset() + return + } + + if p.inFlight == 0 && n < p.f.staged.inlineBelow { + // The writer is idle with nothing in flight, so the listener may commit with its context and flush the tun + p.metricInline.Inc(1) + p.process(b, p.wrxc) + b.reset() + return + } + + parts := min(len(b.chunks), b.data/rxChunkMin) + p.metricBatches.Inc(1) + if parts > 0 { + size := (n + parts - 1) / parts + for off := 0; off < n; off += size { + c := &b.chunks[b.nChunks] + c.items = b.items[off:min(off+size, n)] + b.nChunks++ + } + } + // The jobs go out before the batch does: once the writer has it, it may finish and reset it at any moment + for i := range b.nChunks { + p.pool.jobs <- &b.chunks[i] + } + p.inFlight++ + p.ordered <- b + + if len(p.free) == 0 { + // Every batch is in flight, wait for the writer: the socket buffer absorbs the backlog meanwhile. The wait is + // downstream backpressure, not listener busy time. + t := time.Now() + p.release(<-p.returned) + p.started = p.started.Add(time.Since(t)) + } + p.cur = p.free[len(p.free)-1] + p.free = p.free[:len(p.free)-1] +} + +// reap takes back every batch the writer has finished with, without waiting. +func (p *stagedRx) reap() { + for { + select { + case b := <-p.returned: + p.release(b) + default: + return + } + } +} + +func (p *stagedRx) release(b *rxBatch) { + p.inFlight-- + p.free = append(p.free, b) +} + +// process is the ordered stage for one batch: decrypt whatever no worker is decrypting, commit every item in arrival +// order, and flush the tun batcher. It reports how long it waited on workers. +func (p *stagedRx) process(b *rxBatch, rxc *rxContext) time.Duration { + var waited time.Duration + if b.nChunks == 0 { + decryptRxItems(b.items, rxc.nb) + p.commit(b.items, rxc) + } else { + // Commit chunk by chunk, so the commit overlaps the workers still on later chunks + for i := range b.nChunks { + c := &b.chunks[i] + select { + case <-c.done: + default: + t := time.Now() + <-c.done + waited += time.Since(t) + } + p.commit(c.items, rxc) + } + } + p.f.flushOutside(rxc) + return waited +} + +// commit runs the ordered stage over items in arrival order. A non-data packet goes through readOutsidePackets in +// its place, a run of data packets on one tunnel through commitRun. +func (p *stagedRx) commit(items []rxItem, rxc *rxContext) { + for i := 0; i < len(items); { + ci := items[i].ci + if ci == nil { + p.f.readOutsidePackets(ViaSender{UdpAddr: items[i].via}, items[i].packet, rxc) + i++ + continue + } + j := i + 1 + for j < len(items) && items[j].ci == ci { + j++ + } + p.commitRun(items[i:j], rxc) + i = j + } +} + +// commitRun commits a run of decrypted data packets on one tunnel. The window updates go first, under one lock, then +// the rest; handling a data packet never touches the window, so that is the same as one packet at a time. +func (p *stagedRx) commitRun(run []rxItem, rxc *rxContext) { + f := p.f + ci := run[0].ci + // Classification is speculative: earlier control packets may have removed this tunnel. Resolve it through the + // writer's batch-local cache, exactly where inline processing would, before touching replay state. + index := binary.BigEndian.Uint32(run[0].packet[4:8]) + hostinfo := f.hostMap.QueryIndexCached(index, rxc.hostmapCache) + if hostinfo != run[0].hostinfo || hostinfo.ConnectionState != ci { + if hostinfo == nil { + for i := range run { + f.maybeSendRecvError(run[i].via, index) + } + } + return + } + + ci.decryptLock.Lock() + for i := range run { + // A packet that failed to decrypt does not touch the window, like ConnectionState.Decrypt + if run[i].err == nil && !ci.window.Update(f.l, run[i].counter) { + run[i].err = ErrAlreadySeen + } + } + ci.decryptLock.Unlock() + for i := range run { + it := &run[i] + if f.finishOutsideDecrypt(it.hostinfo, ViaSender{UdpAddr: it.via}, it.counter, it.err) { + f.handleOutsideMessagePacket(it.hostinfo, it.counter, it.out, rxc) + } + } +} + +// writer is the ordered receive stage: window, firewall, coalescer and tun write, one batch at a time in arrival +// order. +func (p *stagedRx) writer() { + defer close(p.writerDone) + for b := range p.ordered { + if p.f.closed.Load() { + // The tun is going away, a write would only log errors. Still wait out the workers on the batch. + for i := range b.nChunks { + <-b.chunks[i].done + } + } else { + t := time.Now() + waited := p.process(b, p.wrxc) + p.writeBusy.Inc(int64(time.Since(t) - waited)) + } + b.reset() + p.returned <- b + } +} + +// close commits anything still in flight, in order, and stops the writer and the workers. +func (p *stagedRx) close() { + if len(p.cur.items) > 0 { + p.endBatch() + } + close(p.ordered) + <-p.writerDone + p.pool.close() +} diff --git a/flow_rx_test.go b/flow_rx_test.go new file mode 100644 index 00000000..36a0f627 --- /dev/null +++ b/flow_rx_test.go @@ -0,0 +1,237 @@ +//go:build linux && !android && !e2e_testing + +package nebula + +import ( + "fmt" + "net/netip" + "sync/atomic" + "testing" + "time" + + "github.com/flynn/noise" + "github.com/rcrowley/go-metrics" + "github.com/slackhq/nebula/header" + "github.com/slackhq/nebula/noiseutil" + "github.com/slackhq/nebula/udp" + "github.com/stretchr/testify/require" + "go.uber.org/goleak" +) + +// TestFlowStagedRXOrderAndReplay feeds every packet behind a forgery of it, from another address, and a replay of +// it, then a test packet that jumps the window and the whole stream again, with shuffled workers and a slow tun, +// through small and large batches: the tun sees each segment once, in order, the window marks them all, the forgeries +// neither consume a counter nor roam, and the jump does not get the earlier data packets of its batch rejected. +func TestFlowStagedRXOrderAndReplay(t *testing.T) { + defer goleak.VerifyNone(t, goleak.IgnoreCurrent()) + for _, cipher := range []noise.CipherFunc{noiseutil.CipherAESGCM, noise.CipherChaChaPoly} { + for _, batchLen := range []int{8, 1000} { + t.Run(fmt.Sprintf("%s/batch=%d", cipher.CipherName(), batchLen), func(t *testing.T) { + sender, receiver := newFlowTestPair(t, cipher, stagedOn(4)) + delayCrypto(receiver, func(n uint64) { + if n%32 == 0 { + time.Sleep(time.Millisecond) + } + }) + receiver.tun.delay = 100 * time.Microsecond + batches, want := flowTestBatches(30, 4, 1, 10) + wire := wireFor(t, sender, batches) + var feed [][]byte + for i, p := range wire { + // Corrupt the tag or the payload + bad := append([]byte(nil), p...) + bad[[]int{len(p) - 1, header.Len + 3}[i%2]] ^= 1 + feed = append(feed, bad, p, p) + } + // The jump lands in a batch with fresh data before it, at batch=1000 + sender.hostinfo.ConnectionState.messageCounter.Add(2 * ReplayWindow) + feed = append(append(feed, sendControl(sender, header.Test, header.TestReply)), wire...) + forged := func(i int) netip.AddrPort { + if i < 3*len(wire) && i%3 == 0 { + return netip.MustParseAddrPort("198.51.100.9:4242") + } + return flowTestUnder + } + rx0 := rxBatches.Count() + conn := &rxConn{wire: feed, batchLen: batchLen, from: forged} + receiver.receive(conn) + require.Positive(t, rxBatches.Count()-rx0) + require.Equal(t, (len(feed)+batchLen-1)/batchLen, conn.swaps, "every batch is handed off") + require.Equal(t, want, receiver.tun.segments(t), "every packet exactly once, in order") + require.Equal(t, flowTestUnder, receiver.hostinfo.GetRemote()) + for _, p := range wire { + h := &header.H{} + require.NoError(t, h.Parse(p)) + require.False(t, receiver.hostinfo.ConnectionState.window.Check(receiver.f.l, h.MessageCounter), "counter %d must be marked seen", h.MessageCounter) + } + }) + } + } +} + +// TestFlowStagedRXClose: a CloseTunnel in the middle of a batch is handled in its place, by the writer. The data after +// it in the batch is still delivered from the batch's hostmap cache, as inline receive does. +func TestFlowStagedRXClose(t *testing.T) { + defer goleak.VerifyNone(t, goleak.IgnoreCurrent()) + sender, receiver := newFlowTestPair(t, noise.CipherChaChaPoly, stagedOn(2)) + batches, want := flowTestBatches(4, 4, 1, 8) + wire := wireFor(t, sender, batches) + half := len(wire) / 2 + feed := append(append(append([][]byte{}, wire[:half]...), sendControl(sender, header.CloseTunnel, 0)), wire[half:]...) + receiver.receive(&rxConn{wire: feed, batchLen: len(feed)}) + require.Equal(t, want, receiver.tun.segments(t)) + require.Nil(t, receiver.f.hostMap.QueryIndex(receiver.hostinfo.localIndexId)) +} + +// sendControl returns the wire packet of a control message from sender. +func sendControl(sender *flowTestPeer, typ header.MessageType, st header.MessageSubType) []byte { + sender.f.sendNoMetrics(typ, st, sender.hostinfo.ConnectionState, sender.hostinfo, flowTestUnder, nil, make([]byte, 12), make([]byte, mtu), 0) + wire := sender.conn.captured() + sender.conn.pkts, sender.conn.addrs = nil, nil + return wire[len(wire)-1] +} + +// receivePaths is inline, staged and, with the threshold, adaptive receive. +func receivePaths() []stagedConfig { + return []stagedConfig{{}, stagedOn(2), adaptive(2)} +} + +// TestFlowStagedRXRecvError feeds a RecvError that deletes the tunnel, then a known replay and new data for it, in one +// batch. Every receive path must drop the data through its own view of the hostmap, without touching the replay window, +// and answer each packet with a RecvError, as inline receive does, without decrypting the replay again. +func TestFlowStagedRXRecvError(t *testing.T) { + for _, sc := range receivePaths() { + t.Run(stagedName(sc), func(t *testing.T) { + sender, receiver := newFlowTestPair(t, noise.CipherChaChaPoly, sc) + receiver.f.messageMetrics = newMessageMetrics() + receiver.f.acceptRecvErrorConfig = recvErrorAlways + receiver.f.sendRecvErrorConfig = recvErrorAlways + batches, want := flowTestBatches(1, 1, 1, 33) + wire := wireFor(t, sender, batches) + rxc := newRxContext(receiver.f, 0) + receiver.f.readOutsidePackets(ViaSender{UdpAddr: flowTestUnder}, append([]byte(nil), wire[0]...), rxc) + receiver.f.flushOutside(rxc) + h := &header.H{} + require.NoError(t, h.Parse(wire[0])) + replay := h.MessageCounter + var decrypts atomic.Int64 + delayCrypto(receiver, func(n uint64) { + if n == replay { + decrypts.Add(1) + } + }) + ctl := header.Encode(make([]byte, header.Len), header.Version, header.RecvError, 0, receiver.hostinfo.remoteIndexId, 0) + conn := &rxConn{wire: append([][]byte{ctl}, wire...), batchLen: len(wire) + 1} + receiver.receive(conn) + require.Nil(t, receiver.f.hostMap.QueryIndex(receiver.hostinfo.localIndexId)) + require.Equal(t, want[:1], receiver.tun.segments(t)) + require.Zero(t, decrypts.Load(), "known replays must not be decrypted again") + replies := conn.captured() + require.Len(t, replies, len(wire)) + for i, reply := range replies { + require.NoError(t, h.Parse(reply)) + require.Equal(t, header.RecvError, h.Type) + require.NoError(t, h.Parse(wire[i])) + require.True(t, i == 0 || receiver.hostinfo.ConnectionState.window.Check(nil, h.MessageCounter), "dropped data must not update replay state") + } + }) + } +} + +// TestFlowStagedRXReplacedTunnel classifies a data packet, then a close and a new handshake replace its tunnel at the +// same index before the writer commits it, as when the listener runs ahead: it must be dropped untouched. +func TestFlowStagedRXReplacedTunnel(t *testing.T) { + sender, receiver := newFlowTestPair(t, noise.CipherChaChaPoly, stagedOn(2)) + receiver.f.messageMetrics = newMessageMetrics() + batches, _ := flowTestBatches(1, 1, 1, 1) + wire := wireFor(t, sender, batches) + closePacket := sendControl(sender, header.CloseTunnel, 0) + p := &stagedRx{f: receiver.f, h: &header.H{}, hostmapCache: map[uint32]*HostInfo{}} + item := rxItem{packet: wire[0], via: flowTestUnder} + p.classify(&item) + require.NotNil(t, item.ci) + rxc := newRxContext(receiver.f, 0) + p.process(&rxBatch{items: []rxItem{{packet: closePacket, via: flowTestUnder}}}, rxc) + require.Nil(t, receiver.f.hostMap.QueryIndex(receiver.hostinfo.localIndexId)) + _, replacement := newFlowTestPair(t, noise.CipherChaChaPoly, stagedOn(2)) + receiver.f.hostMap.Lock() + receiver.f.hostMap.unlockedAddHostInfo(replacement.hostinfo, receiver.f) + receiver.f.hostMap.Unlock() + p.process(&rxBatch{items: []rxItem{item}}, rxc) + require.Empty(t, receiver.tun.writes) + require.True(t, item.ci.window.Check(nil, item.counter)) + require.True(t, replacement.hostinfo.ConnectionState.window.Check(nil, item.counter)) +} + +// TestFlowStagedRXRealSocket exercises buffer handoff through the real Linux UDP receive loop. +func TestFlowStagedRXRealSocket(t *testing.T) { + defer goleak.VerifyNone(t, goleak.IgnoreCurrent()) + sender, receiver := newFlowTestPair(t, noise.CipherChaChaPoly, stagedOn(3)) + batches, want := flowTestBatches(60, 4, 1, 8) + wire := wireFor(t, sender, batches) + + l := receiver.f.l + conn, err := udp.NewListener(l, udp.Settings{Listen: netip.MustParseAddrPort("127.0.0.1:0"), Batch: 16, Offloads: true}) + require.NoError(t, err) + addr, err := conn.LocalAddr() + require.NoError(t, err) + out, err := udp.NewListener(l, udp.Settings{Listen: netip.MustParseAddrPort("127.0.0.1:0"), Batch: 16, Offloads: true}) + require.NoError(t, err) + defer out.Close() + // The sender's address is the tunnel's remote, so no roaming + from, err := out.LocalAddr() + require.NoError(t, err) + receiver.hostinfo.remote.Store(&from) + + done := make(chan struct{}) + go func() { receiver.receive(conn); close(done) }() + // Send in bursts small enough for the default socket buffer + for off := 0; off < len(wire); off += 32 { + end := min(off+32, len(wire)) + addrs := make([]netip.AddrPort, end-off) + for i := range addrs { + addrs[i] = addr + } + _, err := out.WriteBatch(wire[off:end], addrs) + require.NoError(t, err) + require.Eventually(t, func() bool { return len(receiver.tun.segments(t)) >= end }, 5*time.Second, 100*time.Microsecond) + } + receiver.f.closed.Store(true) + require.NoError(t, conn.Close()) + <-done + require.Equal(t, want, receiver.tun.segments(t)) +} + +// TestFlowStagedRXCloseFlushesPartial: packets the listener holds in a batch it never got to end when the conn fails +// are still committed, in order, on a live interface. +func TestFlowStagedRXCloseFlushesPartial(t *testing.T) { + defer goleak.VerifyNone(t, goleak.IgnoreCurrent()) + sender, receiver := newFlowTestPair(t, noise.CipherChaChaPoly, stagedOn(2)) + batches, want := flowTestBatches(11, 4, 1, 8) + wire := wireFor(t, sender, batches) + p := newStagedRx(receiver.f, 0, newRxContext(receiver.f, 0)) + for i, packet := range wire { + p.deliver(flowTestUnder, packet) + if (i+1)%64 == 0 { + p.endBatch() + } + } + p.close() + require.Equal(t, want, receiver.tun.segments(t)) +} + +// TestFlowStagedRXInlineBelow: small idle batches stay on the listener, larger batches still use the ordered writer. +func TestFlowStagedRXInlineBelow(t *testing.T) { + defer goleak.VerifyNone(t, goleak.IgnoreCurrent()) + sc := adaptive(3) + sc.inlineBelow = 16 + sender, receiver := newFlowTestPair(t, noise.CipherChaChaPoly, sc) + batches, want := flowTestBatches(40, 4, 1, 16) + wire := wireFor(t, sender, batches) + inline := metrics.GetOrRegisterCounter("flow.rx.inline", nil) + inline0, rx0 := inline.Count(), rxBatches.Count() + receiver.receive(&rxConn{wire: wire, sizes: []int{8, 8, 8, 128, 128, 8, 8, 128, 4, 128}}) + require.Equal(t, want, receiver.tun.segments(t)) + require.Positive(t, inline.Count()-inline0) + require.Positive(t, rxBatches.Count()-rx0) +} diff --git a/flow_test.go b/flow_test.go new file mode 100644 index 00000000..2553bb01 --- /dev/null +++ b/flow_test.go @@ -0,0 +1,491 @@ +//go:build linux && !android && !e2e_testing + +package nebula + +import ( + "encoding/binary" + "io" + "math" + "net/netip" + "os" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/flynn/noise" + "github.com/gaissmai/bart" + "github.com/rcrowley/go-metrics" + "github.com/slackhq/nebula/cert" + "github.com/slackhq/nebula/config" + "github.com/slackhq/nebula/firewall" + "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/test" + "github.com/slackhq/nebula/udp" + "github.com/stretchr/testify/require" + "go.uber.org/goleak" +) + +// flowTestMSS is the segment payload size of the synthetic TSO superpackets. +const flowTestMSS = 1200 + +var ( + flowTestLocal = netip.MustParseAddr("10.0.0.1") + flowTestRemote = netip.MustParseAddr("10.0.0.2") + flowTestUnder = netip.MustParseAddrPort("192.0.2.2:4242") +) + +// flowTCPSuperpacket builds an IPv4/TCP TSO superpacket from local to remote carrying segs segments. Segment i's +// payload starts with the flow id, the superpacket seq and i, so a receiver can check what it got and in which order. +func flowTCPSuperpacket(flow uint16, seq uint32, segs int) tio.Packet { + const hdr = 40 + pay := segs * flowTestMSS + b := make([]byte, hdr+pay) + b[0] = 0x45 + binary.BigEndian.PutUint16(b[2:4], uint16(min(hdr+pay, 65535))) + b[8] = 64 + b[9] = iputil.IPProtocolTCP + copy(b[12:16], flowTestLocal.AsSlice()) + copy(b[16:20], flowTestRemote.AsSlice()) + binary.BigEndian.PutUint16(b[20:22], 10000+flow) + binary.BigEndian.PutUint16(b[22:24], 80) + binary.BigEndian.PutUint32(b[24:28], seq*uint32(pay)) + b[32] = 0x50 + b[33] = 0x10 // ACK + binary.BigEndian.PutUint16(b[34:36], 65535) + for i := range segs { + p := b[hdr+i*flowTestMSS:] + binary.BigEndian.PutUint16(p[0:2], flow) + binary.BigEndian.PutUint32(p[2:6], seq) + binary.BigEndian.PutUint16(p[6:8], uint16(i)) + } + if segs == 1 { + return tio.Packet{Bytes: b} + } + return tio.Packet{Bytes: b, GSO: tio.GSOInfo{Size: flowTestMSS, HdrLen: hdr, CsumStart: 20, Proto: tio.GSOProtoTCP}} +} + +// decodedSeg identifies one segment by the marker flowTCPSuperpacket put in its payload. +type decodedSeg struct { + flow uint16 + seq uint32 + idx uint16 +} + +func decodeSeg(p []byte) decodedSeg { + return decodedSeg{flow: binary.BigEndian.Uint16(p[0:2]), seq: binary.BigEndian.Uint32(p[2:6]), idx: binary.BigEndian.Uint16(p[6:8])} +} + +// flowTestBatches builds n reads of per superpackets of segs segments, across flows interleaved flows, and the +// segments they carry in order. +func flowTestBatches(n, per, flows, segs int) ([][]tio.Packet, []decodedSeg) { + var batches [][]tio.Packet + var want []decodedSeg + seq := make([]uint32, flows) + for range n { + var b []tio.Packet + for j := range per { + flow := uint16(j % flows) + b = append(b, flowTCPSuperpacket(flow, seq[flow], segs)) + for s := range segs { + want = append(want, decodedSeg{flow: flow, seq: seq[flow], idx: uint16(s)}) + } + seq[flow]++ + } + batches = append(batches, b) + } + return batches, want +} + +// flowTestPeer is one end of a tunnel: the interface, the hostinfo for the other end, and its capture points. +type flowTestPeer struct { + f *Interface + hostinfo *HostInfo + conn *flowCaptureConn + tun *flowTunCapture +} + +// newFlowTestPair builds two interfaces joined by one real tunnel (keys from an actual handshake). sender sends from +// flowTestLocal to flowTestRemote, receiver is the far side. +func newFlowTestPair(t testing.TB, cipher noise.CipherFunc, sc stagedConfig) (sender, receiver *flowTestPeer) { + t.Helper() + initR, respR := runTestHandshakeWith(t, cipher) + sci, err := newConnectionStateFromResult(initR) + require.NoError(t, err) + rci, err := newConnectionStateFromResult(respR) + require.NoError(t, err) + sender = newFlowTestPeer(t, flowTestLocal, flowTestRemote, sci, 100, 200) + receiver = newFlowTestPeer(t, flowTestRemote, flowTestLocal, rci, 200, 100) + sender.f.staged, receiver.f.staged = sc, sc + return sender, receiver +} + +func newFlowTestPeer(t testing.TB, me, them netip.Addr, ci *ConnectionState, localIdx, remoteIdx uint32) *flowTestPeer { + l := test.NewLogger() + hm := newHostMap(l) + hm.preferredRanges.Store(&[]netip.Prefix{}) + + myNets, myAddrs := &bart.Lite{}, &bart.Lite{} + myNets.Insert(netip.PrefixFrom(me, 24)) + myAddrs.Insert(netip.PrefixFrom(me, 32)) + myCert := &dummyCert{version: cert.Version1, networks: []netip.Prefix{netip.PrefixFrom(me, 24)}} + theirCert := &dummyCert{version: cert.Version1, networks: []netip.Prefix{netip.PrefixFrom(them, 24)}, name: "peer"} + ci.peerCert = &cert.CachedCertificate{Certificate: theirCert, InvertedGroups: map[string]struct{}{}} + + fw := NewFirewall(l, time.Minute, time.Minute, time.Minute, myCert) + require.NoError(t, fw.AddRule(false, firewall.ProtoAny, 0, 0, []string{"any"}, "", "", "", "", "")) + require.NoError(t, fw.AddRule(true, firewall.ProtoAny, 0, 0, []string{"any"}, "", "", "", "", "")) + + conn, tun := &flowCaptureConn{}, &flowTunCapture{} + lh := newTestLighthouse() + lh.remoteAllowList.Store(&RemoteAllowList{}) + + f := &Interface{ + hostMap: hm, + outside: conn, + firewall: fw, + lightHouse: lh, + pki: &PKI{}, + myVpnNetworksTable: myNets, + myVpnAddrsTable: myAddrs, + routines: 1, + writers: []udp.Conn{conn}, + messageMetrics: &MessageMetrics{txExhausted: metrics.NewCounter()}, + metricTxDropped: metrics.NilCounter{}, + l: l, + } + f.ctx = t.Context() + f.pki.caPool.Store(cert.NewCAPool()) + // Never promote or requery, those would put extra test packets on the wire + f.tryPromoteEvery.Store(math.MaxUint32) + f.reQueryEvery.Store(math.MaxUint32) + f.batchers = []*batch.MultiCoalescer{batch.NewMultiCoalescer(tun, l)} + f.connectionManager = &connectionManager{ + intf: f, + relayUsed: map[uint32]struct{}{}, + relayUsedLock: &sync.RWMutex{}, + trafficTimer: NewLockingTimerWheel[uint32](time.Second, time.Minute), + } + f.handshakeManager = NewHandshakeManager(l, hm, lh, conn, defaultHandshakeConfig) + f.handshakeManager.f = f + + hostinfo := &HostInfo{ + ConnectionState: ci, + vpnAddrs: []netip.Addr{them}, + localIndexId: localIdx, + remoteIndexId: remoteIdx, + relayState: RelayState{relayForByAddr: map[netip.Addr]*Relay{}, relayForByIdx: map[uint32]*Relay{}}, + } + hostinfo.remote.Store(&flowTestUnder) + hostinfo.buildNetworks(myNets, theirCert) + hm.unlockedAddHostInfo(hostinfo, f) + + return &flowTestPeer{f: f, hostinfo: hostinfo, conn: conn, tun: tun} +} + +// flowCaptureConn records every packet sent, in order, as copies. +type flowCaptureConn struct { + udp.NoopConn + mu sync.Mutex + pkts [][]byte + addrs []netip.AddrPort + // delay, when set, slows every WriteBatch down so batches pile up in flight + delay time.Duration +} + +func (c *flowCaptureConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort) (int, error) { + time.Sleep(c.delay) + c.mu.Lock() + defer c.mu.Unlock() + for i, b := range bufs { + c.pkts = append(c.pkts, append([]byte(nil), b...)) + c.addrs = append(c.addrs, addrs[i]) + } + return len(bufs), nil +} + +func (c *flowCaptureConn) WriteTo(b []byte, addr netip.AddrPort) error { + _, err := c.WriteBatch([][]byte{b}, []netip.AddrPort{addr}) + return err +} + +func (c *flowCaptureConn) captured() [][]byte { + c.mu.Lock() + defer c.mu.Unlock() + return append([][]byte(nil), c.pkts...) +} + +// flowTunCapture is a tun queue that records every plaintext write. +type flowTunCapture struct { + mu sync.Mutex + writes [][]byte + // delay, when set, slows every write down + delay time.Duration +} + +func (c *flowTunCapture) Read() ([]tio.Packet, error) { return nil, io.EOF } +func (c *flowTunCapture) Close() error { return nil } +func (c *flowTunCapture) Write(b []byte) (int, error) { + time.Sleep(c.delay) + c.mu.Lock() + defer c.mu.Unlock() + c.writes = append(c.writes, append([]byte(nil), b...)) + return len(b), nil +} + +// segments decodes what reached the tun. The coalescer may have merged segments back into superpackets, so this +// walks each payload in MSS steps. +func (c *flowTunCapture) segments(t testing.TB) []decodedSeg { + t.Helper() + c.mu.Lock() + defer c.mu.Unlock() + var out []decodedSeg + for _, w := range c.writes { + pay := w[40:] + require.Zero(t, len(pay)%flowTestMSS, "tun write payload %d bytes", len(pay)) + for off := 0; off < len(pay); off += flowTestMSS { + out = append(out, decodeSeg(pay[off:])) + } + } + return out +} + +// decryptWire checks every wire packet decrypts on the receiver, in order, with strictly increasing counters, and +// returns what each one carried. It uses its own replay window and copies, so the wire can still be fed to the +// receiver afterwards. +func decryptWire(t testing.TB, receiver *flowTestPeer, wire [][]byte) []decodedSeg { + t.Helper() + peer := receiver.hostinfo.ConnectionState + ci := &ConnectionState{dKey: peer.dKey, window: NewBits(peer.window.length)} + nb := make([]byte, 12) + var last uint64 + out := make([]decodedSeg, 0, len(wire)) + for i, pkt := range wire { + h := &header.H{} + require.NoError(t, h.Parse(pkt)) + require.Equal(t, header.Message, h.Type) + require.Equal(t, receiver.hostinfo.localIndexId, h.RemoteIndex) + require.Greater(t, h.MessageCounter, last, "wire packet %d: counter %d after %d", i, h.MessageCounter, last) + last = h.MessageCounter + plain, err := ci.Decrypt(receiver.f.l, h.MessageCounter, append([]byte(nil), pkt...), nb) + require.NoError(t, err, "wire packet %d", i) + out = append(out, decodeSeg(plain[40:])) + } + return out +} + +// flowBatchQueue is a tun queue that hands out scripted Read batches, then fails with io.EOF. Every batch is copied +// into the current receive buffer, so a test sees the buffer ownership the Offload queue has. +type flowBatchQueue struct { + batches [][]tio.Packet + rxBuf []byte + out []tio.Packet + closed chan struct{} + // reads counts the batches handed out + reads atomic.Int64 +} + +func newFlowBatchQueue(batches [][]tio.Packet) *flowBatchQueue { + return &flowBatchQueue{batches: batches, rxBuf: make([]byte, 8*65536), closed: make(chan struct{})} +} + +func (q *flowBatchQueue) Read() ([]tio.Packet, error) { + select { + case <-q.closed: + return nil, os.ErrClosed + default: + } + n := int(q.reads.Load()) + if n == len(q.batches) { + return nil, io.EOF + } + q.out = q.out[:0] + off := 0 + for _, p := range q.batches[n] { + l := copy(q.rxBuf[off:], p.Bytes) + q.out = append(q.out, tio.Packet{Bytes: q.rxBuf[off : off+l : off+l], GSO: p.GSO}) + off += l + } + q.reads.Add(1) + return q.out, nil +} + +func (q *flowBatchQueue) Write(b []byte) (int, error) { return len(b), nil } + +func (q *flowBatchQueue) Close() error { close(q.closed); return nil } + +// delayedCipher changes scheduling through the real cipher interface, only in tests. +type delayedCipher struct { + noiseutil.CipherState + before func(uint64) +} + +func (c delayedCipher) EncryptDanger(out, ad, plain []byte, n uint64, nb []byte) ([]byte, error) { + c.before(n) + return c.CipherState.EncryptDanger(out, ad, plain, n, nb) +} + +func (c delayedCipher) DecryptDanger(out, ad, ciphertext []byte, n uint64, nb []byte) ([]byte, error) { + c.before(n) + return c.CipherState.DecryptDanger(out, ad, ciphertext, n, nb) +} + +func delayCrypto(peer *flowTestPeer, before func(uint64)) { + ci := peer.hostinfo.ConnectionState + ci.eKey = delayedCipher{ci.eKey, before} + ci.dKey = delayedCipher{ci.dKey, before} +} + +// stagedOn enables staging with n workers. +func stagedOn(n int) stagedConfig { + return stagedConfig{workers: n} +} + +// adaptive is staging with the inline threshold parseStagedConfig sets. +func adaptive(n int) stagedConfig { + return stagedConfig{workers: n, inlineBelow: 256} +} + +func stagedName(sc stagedConfig) string { + switch { + case sc.workers == 0: + return "inline" + case sc.inlineBelow > 0: + return "adaptive" + default: + return "staged" + } +} + +// TestFlowStagedFallback: a conn that cannot hand off its batches keeps receive on the inline path. +func TestFlowStagedFallback(t *testing.T) { + defer goleak.VerifyNone(t, goleak.IgnoreCurrent()) + sender, receiver := newFlowTestPair(t, noise.CipherChaChaPoly, stagedOn(3)) + batches, want := flowTestBatches(10, 2, 1, 4) + wire := wireFor(t, sender, batches) + require.Equal(t, want, decryptWire(t, receiver, wire)) + rx0 := rxBatches.Count() + receiver.receive(plainConn{&rxConn{wire: wire, batchLen: 16}}) + require.Equal(t, want, receiver.tun.segments(t)) + require.Equal(t, rx0, rxBatches.Count()) +} + +// wireFor produces the wire packets the sender emits for batches. +func wireFor(t testing.TB, sender *flowTestPeer, batches [][]tio.Packet) [][]byte { + t.Helper() + sender.f.listenIn(newFlowBatchQueue(batches), 0) + wire := sender.conn.captured() + sender.conn.pkts, sender.conn.addrs = nil, nil + return wire +} + +// rxConn plays wire back to listenOut through the buffer handoff, in batches of batchLen or of the scripted sizes +// (the last one repeating), and captures what is sent back. Every buffer handed back is scribbled over before it is +// read into, so a stage still using one shows corruption. from, when set, picks each packet's source address by its +// index in wire. +type rxConn struct { + flowCaptureConn + wire [][]byte + batchLen int + sizes []int + from func(i int) netip.AddrPort + swaps int +} + +func (c *rxConn) ListenOut(r udp.EncReader, flush func()) error { + return c.ListenOutHandoff(r, func(bufs [][]byte) [][]byte { + flush() + return bufs + }) +} + +func (c *rxConn) ListenOutHandoff(r udp.EncReader, handoff func(bufs [][]byte) [][]byte) error { + n := c.batchLen + for _, size := range c.sizes { + n = max(n, size) + } + bufs := make([][]byte, n) + for i := range bufs { + bufs[i] = make([]byte, udp.MTU) + } + for off, k := 0, 0; off < len(c.wire); k++ { + size := c.batchLen + if len(c.sizes) > 0 { + size = c.sizes[min(k, len(c.sizes)-1)] + } + end := min(off+size, len(c.wire)) + for i, p := range c.wire[off:end] { + n := copy(bufs[i], p) + from := flowTestUnder + if c.from != nil { + from = c.from(off + i) + } + r(from, bufs[i][:n:n]) + } + next := handoff(bufs) + if len(next) != len(bufs) { + panic("handoff returned the wrong number of buffers") + } + bufs = next + for _, b := range bufs { + for j := range b[:udp.MTU] { + b[j] = 0x5a + } + } + c.swaps++ + off = end + } + return os.ErrClosed +} + +// plainConn hides ListenOutHandoff, so it cannot hand off its batches. +type plainConn struct{ udp.Conn } + +var rxBatches = metrics.GetOrRegisterCounter("flow.rx.batches", nil) + +// receive runs the peer's listenOut over conn until conn runs out. +func (p *flowTestPeer) receive(conn udp.Conn) { + p.f.writers = []udp.Conn{conn} + p.f.outside = conn + p.f.listenOut(0) +} + +// TestFlowStagedFirewallParity: both receive paths apply the same inbound policy. An authenticated inbound denial +// still consumes its replay counter, exactly as inline Decrypt does. +func TestFlowStagedFirewallParity(t *testing.T) { + for _, sc := range []stagedConfig{{}, stagedOn(2)} { + t.Run(stagedName(sc), func(t *testing.T) { + batches, _ := flowTestBatches(4, 3, 1, 8) + sender, receiver := newFlowTestPair(t, noise.CipherChaChaPoly, sc) + wire := wireFor(t, sender, batches) + receiver.f.firewall.InRules = newFirewallTable() + receiver.receive(&rxConn{wire: wire, batchLen: 64}) + require.Empty(t, receiver.tun.writes) + for _, pkt := range wire { + h := &header.H{} + require.NoError(t, h.Parse(pkt)) + require.False(t, receiver.hostinfo.ConnectionState.window.Check(receiver.f.l, h.MessageCounter)) + } + }) + } +} + +// TestFlowStagedAESGCM runs the parsed configuration with the real AES cipher. Receive decryption has no nonce +// ordering requirement, so it stays staged in BoringCrypto and FIPS builds too. +func TestFlowStagedAESGCM(t *testing.T) { + l := test.NewLogger() + c := config.NewC(l) + c.Settings = map[string]any{"tun": map[string]any{"staged": map[string]any{"workers": 2}}} + sender, receiver := newFlowTestPair(t, noiseutil.CipherAESGCM, parseStagedConfig(c, l)) + require.Equal(t, 2, receiver.f.staged.workers) + batches, want := flowTestBatches(4, 8, 1, 45) + wire := wireFor(t, sender, batches) + rx0 := rxBatches.Count() + receiver.receive(&rxConn{wire: wire, batchLen: 512}) + require.Greater(t, rxBatches.Count(), rx0) + require.Equal(t, want, receiver.tun.segments(t)) +} diff --git a/interface.go b/interface.go index 29051be4..bd18ce4e 100644 --- a/interface.go +++ b/interface.go @@ -65,6 +65,7 @@ type InterfaceConfig struct { // goroutine's sendmmsg on one XPS-selected NIC TX ring so per-flow // packets stay ordered on the wire. PinThreads bool + staged stagedConfig l *slog.Logger } @@ -99,6 +100,7 @@ type Interface struct { // a CPU at all (tun.pin_threads, default true). When false, threads are // left free to migrate as on stock nebula. pinThreads bool + staged stagedConfig relayManager *relayManager tryPromoteEvery atomic.Uint32 @@ -248,6 +250,7 @@ func NewInterface(ctx context.Context, c *InterfaceConfig) (*Interface, error) { conntrackCacheTimeout: c.ConntrackCacheTimeout, cpuAffinity: c.CpuAffinity, pinThreads: c.PinThreads, + staged: c.staged, udpTxErrLimiter: udpTxErrLimiter, udpRelayTxErrLimiter: udpRelayTxErrLimiter, txOversizeLimiter: txOversizeLimiter, @@ -413,14 +416,18 @@ func (f *Interface) listenOut(i int) { f.readOutsidePackets(ViaSender{UdpAddr: fromUdpAddr}, payload, rxc) } - flusher := func() { - if err := f.batchers[i].Flush(); err != nil { - f.l.Error("Failed to flush tun coalescer", "error", err) - } - clear(rxc.hostmapCache) - } + flusher := func() { f.flushOutside(rxc) } - err := li.ListenOut(listener, flusher) + var err error + if hc, ok := li.(handoffConn); ok && f.staged.workers > 0 { + // Every stage on its own goroutine, see flow_rx.go. rxc becomes the ordered writer's context. + err = newStagedRx(f, i, rxc).listen(hc) + } else { + if f.staged.workers > 0 { + f.l.Info("udp conn cannot hand off receive batches, receive stays inline", "reader", i) + } + err = li.ListenOut(listener, flusher) + } // An error after teardown began is shutdown noise, the closed flag covers resources // Close releases itself and the cancelled ctx covers ones torn down by their owners diff --git a/main.go b/main.go index a8d01779..ec200d4d 100644 --- a/main.go +++ b/main.go @@ -244,6 +244,7 @@ func Main(c *config.C, configTest bool, buildVersion string, l *slog.Logger, dev pinThreads := c.GetBool("tun.pin_threads", true) cpuAffinity := parseCpuAffinity(c, l, routines) + staged := parseStagedConfig(c, l) if pinThreads && routines > 1 && len(cpuAffinity) == 0 && !configTest { // The operator didn't choose pin CPUs, so pick a default set that // prefers performance cores and doesn't stack co-located instances @@ -296,6 +297,7 @@ func Main(c *config.C, configTest bool, buildVersion string, l *slog.Logger, dev ConntrackCacheTimeout: conntrackCacheTimeout, CpuAffinity: cpuAffinity, PinThreads: pinThreads, + staged: staged, l: l, } diff --git a/main_test.go b/main_test.go index e217ffbf..43274ecb 100644 --- a/main_test.go +++ b/main_test.go @@ -5,6 +5,7 @@ import ( "net/netip" "os" "path/filepath" + "runtime" "testing" "time" @@ -123,3 +124,23 @@ firewall: require.NoError(t, err) require.Nil(t, ctrl, "a config test hands back nothing to stop, so it must stop itself") } + +func TestParseStagedConfig(t *testing.T) { + l := test.NewLogger() + staged := func(workers any) stagedConfig { + c := config.NewC(l) + if workers != nil { + c.Settings["tun"] = map[string]any{"staged": map[string]any{"workers": workers}} + } + return parseStagedConfig(c, l) + } + for _, workers := range []any{nil, -1, 0} { + require.Equal(t, stagedConfig{}, staged(workers)) + } + if runtime.GOOS != "linux" { + require.Equal(t, stagedConfig{}, staged(2)) + return + } + want := stagedConfig{workers: 2, inlineBelow: 256} + require.Equal(t, want, staged(2)) +} diff --git a/outside.go b/outside.go index dc019e36..1bde9b2c 100644 --- a/outside.go +++ b/outside.go @@ -130,17 +130,10 @@ func (f *Interface) readOutsidePackets(via ViaSender, packet []byte, rxc *rxCont } out, err := hostinfo.ConnectionState.Decrypt(f.l, h.MessageCounter, packet, rxc.nb) - if err != nil { - if f.l.Enabled(context.Background(), slog.LevelDebug) { - hostinfo.logger(f.l).Debug("Failed to decrypt packet", "error", err, "from", via, "header", h) - } + if !f.finishOutsideDecrypt(hostinfo, via, h.MessageCounter, err) { return } - // Roam before we respond - f.handleHostRoaming(hostinfo, via) - f.connectionManager.In(hostinfo) - switch h.Type { case header.Message: switch h.Subtype { @@ -473,6 +466,29 @@ func parseV4(data []byte, incoming bool, fp *firewall.ParsedPacket) error { return nil } +// finishOutsideDecrypt logs a failed decrypt, or roams and records the traffic for an authenticated packet before it +// is handled. It reports whether the packet authenticated. +func (f *Interface) finishOutsideDecrypt(hostinfo *HostInfo, via ViaSender, counter uint64, err error) bool { + if err != nil { + if f.l.Enabled(context.Background(), slog.LevelDebug) { + hostinfo.logger(f.l).Debug("Failed to decrypt packet", "error", err, "from", via, "counter", counter) + } + return false + } + // Roam before we respond + f.handleHostRoaming(hostinfo, via) + f.connectionManager.In(hostinfo) + return true +} + +// flushOutside ends a receive batch: it writes out the tun coalescer and drops the batch's hostmap cache. +func (f *Interface) flushOutside(rxc *rxContext) { + if err := f.batchers[rxc.q].Flush(); err != nil { + f.l.Error("Failed to flush tun coalescer", "error", err) + } + clear(rxc.hostmapCache) +} + func (f *Interface) handleOutsideMessagePacket(hostinfo *HostInfo, messageCounter uint64, out []byte, rxc *rxContext) { err := newPacket(out, true, rxc.fwPacket) if err != nil { diff --git a/udp/udp_linux.go b/udp/udp_linux.go index cc834131..6eecdc8e 100644 --- a/udp/udp_linux.go +++ b/udp/udp_linux.go @@ -246,6 +246,18 @@ func getFrom(names [][]byte, i int, isV4 bool) netip.AddrPort { } func (u *StdConn) ListenOut(r EncReader, flush func()) error { + return u.listenOut(r, flush, nil) +} + +// ListenOutHandoff is ListenOut for a reader that keeps the packets: after each batch, handoff takes the batch's +// buffers and returns replacements, of the same count and at least the same size, to receive the next one into. +func (u *StdConn) ListenOutHandoff(r EncReader, handoff func(bufs [][]byte) [][]byte) error { + return u.listenOut(r, nil, handoff) +} + +// listenOut is the receive loop. After each batch it calls flush, or with handoff set, hands the batch's buffers +// over and points the recvmmsg iovecs at the ones it gets back. +func (u *StdConn) listenOut(r EncReader, flush func(), handoff func(bufs [][]byte) [][]byte) error { bufSize := MTU cmsgSpace := 0 if u.groSupported { @@ -283,7 +295,21 @@ func (u *StdConn) ListenOut(r EncReader, flush func()) error { deliverSegments(r, from, payload, segSize) } - flush() + if handoff == nil { + flush() + continue + } + next := handoff(buffers) + if len(next) != len(buffers) { + panic(fmt.Sprintf("udp: handoff returned %d buffers, need %d", len(next), len(buffers))) + } + for i := range next { + if len(next[i]) < bufSize { + panic(fmt.Sprintf("udp: handoff buffer is %d bytes, need %d", len(next[i]), bufSize)) + } + msgs[i].Hdr.Iov.Base = &next[i][0] + } + buffers = next } } diff --git a/udp/udp_tester.go b/udp/udp_tester.go index db90b370..557d4584 100644 --- a/udp/udp_tester.go +++ b/udp/udp_tester.go @@ -183,15 +183,24 @@ func (u *TesterConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort) (int, err } func (u *TesterConn) ListenOut(r EncReader, flush func()) error { + // The batcher borrows plaintext decrypted in place inside p.Data + // until Flush, so the packet must stay alive across flush() + return u.listenOut(r, func(p *Packet) { flush() }) +} + +// ListenOutHandoff hands each packet's buffer over, the replacement goes back to the pool. +func (u *TesterConn) ListenOutHandoff(r EncReader, handoff func(bufs [][]byte) [][]byte) error { + return u.listenOut(r, func(p *Packet) { p.Data = handoff([][]byte{p.Data})[0] }) +} + +func (u *TesterConn) listenOut(r EncReader, finish func(*Packet)) error { for { select { case <-u.done: return os.ErrClosed case p := <-u.RxPackets: r(p.From, p.Data[:len(p.Data):len(p.Data)]) - // The batcher borrows plaintext decrypted in place inside p.Data - // until Flush, so the packet must stay alive across flush() - flush() + finish(p) p.Release() } }