diff --git a/connection_manager_test.go b/connection_manager_test.go index 96d583ad..7de5dbe2 100644 --- a/connection_manager_test.go +++ b/connection_manager_test.go @@ -8,9 +8,13 @@ import ( "testing" "time" + "github.com/flynn/noise" "github.com/rcrowley/go-metrics" "github.com/slackhq/nebula/cert" "github.com/slackhq/nebula/config" + "github.com/slackhq/nebula/header" + "github.com/slackhq/nebula/noiseutil" + "github.com/slackhq/nebula/overlay/batch" "github.com/slackhq/nebula/overlay/overlaytest" "github.com/slackhq/nebula/test" "github.com/slackhq/nebula/udp" @@ -607,6 +611,9 @@ func newPunchTestConnectionManager(t *testing.T, conn udp.Conn) (*connectionMana nc := newConnectionManagerFromConfig(l, config.NewC(l), hostMap, punchy) nc.intf = ifce ifce.connectionManager = nc + // Tunnel sends go through conn too, though without a key from withTestKey they never get as far as writing + ifce.writers = []udp.Conn{conn} + ifce.metricTxDropped = metrics.NilCounter{} hostinfo := newPunchTarget() hostinfo.localIndexId = 1099 @@ -657,6 +664,13 @@ func Test_sendPunch_probesWhenRemoteUnreachable(t *testing.T) { nc.sendPunch(hostinfo) assert.Equal(t, []netip.AddrPort{punchTargetV6}, conn.wrote) assert.NotZero(t, hostinfo.nextUnreachableProbe.Load(), "the tunnel should have probed its other remotes") + assert.Equal(t, []netip.Addr{hostinfo.vpnAddrs[0]}, drainQueries(nc.intf.lightHouse), + "the probe should ask the lighthouse to have the peer punch us") + + // Within the probe interval nothing more goes out, however many sends fail + nc.sendPunch(hostinfo) + nc.sendPunch(hostinfo) + assert.Empty(t, drainQueries(nc.intf.lightHouse)) }) t.Run("delivered", func(t *testing.T) { @@ -666,6 +680,7 @@ func Test_sendPunch_probesWhenRemoteUnreachable(t *testing.T) { nc.sendPunch(hostinfo) assert.Equal(t, []netip.AddrPort{punchTargetV6}, conn.wrote) assert.Zero(t, hostinfo.nextUnreachableProbe.Load()) + assert.Empty(t, drainQueries(nc.intf.lightHouse)) }) t.Run("other failure", func(t *testing.T) { @@ -674,5 +689,133 @@ func Test_sendPunch_probesWhenRemoteUnreachable(t *testing.T) { nc.sendPunch(hostinfo) assert.Zero(t, hostinfo.nextUnreachableProbe.Load(), "only a missing route says to look elsewhere") + assert.Empty(t, drainQueries(nc.intf.lightHouse)) }) } + +// A packet the data path had no route for names its tunnel in its header, the peer's index for it +func Test_sendUnreachable_findsTunnelFromHeader(t *testing.T) { + head := func(st header.MessageSubType, remoteIndex uint32) []byte { + return header.Encode(make([]byte, header.Len), header.Version, header.Message, st, remoteIndex, 1) + } + + t.Run("direct", func(t *testing.T) { + nc, hostinfo := newPunchTestConnectionManager(t, &recordingConn{}) + nc.intf.sendUnreachable(head(header.MessageNone, hostinfo.remoteIndexId), punchTargetV6) + assert.NotZero(t, hostinfo.nextUnreachableProbe.Load(), "the tunnel should have probed its other remotes") + assert.Equal(t, []netip.Addr{hostinfo.vpnAddrs[0]}, drainQueries(nc.intf.lightHouse)) + }) + + t.Run("not its remote", func(t *testing.T) { + nc, hostinfo := newPunchTestConnectionManager(t, &recordingConn{}) + nc.intf.sendUnreachable(head(header.MessageNone, hostinfo.remoteIndexId), punchTargetV4) + assert.Zero(t, hostinfo.nextUnreachableProbe.Load(), "a failure on a remote the tunnel is not using is stale") + }) + + t.Run("relayed", func(t *testing.T) { + nc, hostinfo := newPunchTestConnectionManager(t, &recordingConn{}) + // A relay header's index is the relay session's, it must not be taken for a tunnel's even when they collide + nc.intf.sendUnreachable(head(header.MessageRelay, hostinfo.remoteIndexId), punchTargetV6) + assert.Zero(t, hostinfo.nextUnreachableProbe.Load()) + assert.Empty(t, drainQueries(nc.intf.lightHouse)) + }) + + t.Run("unknown tunnel", func(t *testing.T) { + nc, hostinfo := newPunchTestConnectionManager(t, &recordingConn{}) + nc.intf.sendUnreachable(head(header.MessageNone, hostinfo.remoteIndexId+1), punchTargetV6) + assert.Zero(t, hostinfo.nextUnreachableProbe.Load()) + }) +} + +// withTestKey gives a tunnel a real key so its sends get encrypted and written instead of returning early +func withTestKey(hostinfo *HostInfo) { + cs := noise.NewCipherSuite(noise.DH25519, noise.CipherAESGCM, noise.HashSHA256) + hostinfo.ConnectionState.eKey = noiseutil.NewCipherState(noise.UnsafeNewCipherState(cs, [32]byte{}, 0), noise.CipherAESGCM) +} + +// The probe itself, end to end from a failed send: a test packet to every remote but the one we cannot reach, and a +// lighthouse query, through both places a send can fail. +func Test_remoteUnreachable_probesOtherRemotes(t *testing.T) { + t.Run("control send", func(t *testing.T) { + conn := &recordingConn{fail: map[netip.AddrPort]error{punchTargetV6: errNoRoute}} + nc, hostinfo := newPunchTestConnectionManager(t, conn) + withTestKey(hostinfo) + + nc.intf.SendMessageToHostInfo(header.Test, header.TestRequest, hostinfo, []byte{}, make([]byte, 12), make([]byte, mtu)) + assert.Equal(t, []netip.AddrPort{punchTargetV6, punchTargetV4}, conn.wrote, "the failed send, then the probe") + assert.Equal(t, []header.MessageType{header.Test, header.Test}, conn.types) + assert.Equal(t, []netip.Addr{hostinfo.vpnAddrs[0]}, drainQueries(nc.intf.lightHouse)) + }) + + t.Run("data path batch", func(t *testing.T) { + conn := &recordingConn{fail: map[netip.AddrPort]error{punchTargetV6: errNoRoute}} + nc, hostinfo := newPunchTestConnectionManager(t, conn) + withTestKey(hostinfo) + + // Two of the tunnel's packets fail in one batch, they probe once + sb := batch.NewSendBatch(conn, 4, 4096) + for c := uint64(1); c <= 2; c++ { + pkt := header.Encode(sb.Reserve(64), header.Version, header.Message, header.MessageNone, hostinfo.remoteIndexId, c) + sb.Commit(append(pkt, make([]byte, 32)...), punchTargetV6) + } + nc.intf.flushSendBatch(sb, 0) + + assert.Equal(t, []netip.AddrPort{punchTargetV6, punchTargetV6, punchTargetV4}, conn.wrote, + "the failed packets, then one probe") + assert.Equal(t, header.Test, conn.types[2]) + assert.Equal(t, []netip.Addr{hostinfo.vpnAddrs[0]}, drainQueries(nc.intf.lightHouse)) + }) + + t.Run("closing", func(t *testing.T) { + conn := &recordingConn{fail: map[netip.AddrPort]error{punchTargetV6: errNoRoute}} + nc, hostinfo := newPunchTestConnectionManager(t, conn) + withTestKey(hostinfo) + + nc.intf.SendMessageToHostInfo(header.CloseTunnel, 0, hostinfo, []byte{}, make([]byte, 12), make([]byte, mtu)) + assert.Equal(t, []netip.AddrPort{punchTargetV6}, conn.wrote, "a tunnel being torn down does not probe") + assert.Empty(t, drainQueries(nc.intf.lightHouse)) + }) + + t.Run("lighthouse", func(t *testing.T) { + conn := &recordingConn{fail: map[netip.AddrPort]error{punchTargetV6: errNoRoute}} + nc, hostinfo := newPunchTestConnectionManager(t, conn) + withTestKey(hostinfo) + nc.intf.lightHouse.amLighthouse = true + + nc.intf.SendMessageToHostInfo(header.Test, header.TestRequest, hostinfo, []byte{}, make([]byte, 12), make([]byte, mtu)) + assert.Equal(t, []netip.AddrPort{punchTargetV6}, conn.wrote, "a lighthouse does not roam its clients") + }) + + t.Run("nothing else to probe", func(t *testing.T) { + conn := &recordingConn{fail: map[netip.AddrPort]error{punchTargetV6: errNoRoute}} + nc, hostinfo := newPunchTestConnectionManager(t, conn) + withTestKey(hostinfo) + nc.intf.reQueryWait.Store(int64(time.Minute)) + // Forget the IPv4 remote, all the tunnel knows is the one it cannot reach + hostinfo.remotes = NewRemoteList(hostinfo.vpnAddrs, nil) + hostinfo.remotes.LearnRemote(hostinfo.vpnAddrs[0], punchTargetV6) + + nc.intf.SendMessageToHostInfo(header.Test, header.TestRequest, hostinfo, []byte{}, make([]byte, 12), make([]byte, mtu)) + assert.Equal(t, []netip.Addr{hostinfo.vpnAddrs[0]}, drainQueries(nc.intf.lightHouse), + "the lighthouse is the only way to learn of another remote") + + // Past the probe interval, but the lighthouse is only asked again on the requery schedule + hostinfo.nextUnreachableProbe.Store(0) + nc.intf.SendMessageToHostInfo(header.Test, header.TestRequest, hostinfo, []byte{}, make([]byte, 12), make([]byte, mtu)) + assert.Empty(t, drainQueries(nc.intf.lightHouse)) + assert.Equal(t, []netip.AddrPort{punchTargetV6, punchTargetV6}, conn.wrote) + }) +} + +// drainQueries returns every lighthouse query waiting to go out +func drainQueries(lh *LightHouse) []netip.Addr { + var queries []netip.Addr + for { + select { + case addr := <-lh.queryChan: + queries = append(queries, addr) + default: + return queries + } + } +} diff --git a/control_tester.go b/control_tester.go index 59c3447b..0bb440c5 100644 --- a/control_tester.go +++ b/control_tester.go @@ -117,6 +117,11 @@ func (c *Control) SetUDPAddr(addr netip.AddrPort) { c.f.outside.(*udp.TesterConn).SetAddr(addr) } +// GetUDPAddrFor returns the underlay address this node sends to dst from, which differs by family for a dual stack node +func (c *Control) GetUDPAddrFor(dst netip.AddrPort) netip.AddrPort { + return c.f.outside.(*udp.TesterConn).GetAddrFor(dst) +} + // SetDualStackUDPAddrs gives this node one underlay address per family, sending from whichever matches the // destination. An invalid AddrPort leaves the node without that family, sends to it fail with ENETUNREACH. Register // any address the node did not start on with the router as well or nothing will route back. diff --git a/e2e/dual_stack_roam_test.go b/e2e/dual_stack_roam_test.go index a8730403..e29f568c 100644 --- a/e2e/dual_stack_roam_test.go +++ b/e2e/dual_stack_roam_test.go @@ -33,6 +33,8 @@ func TestDualStackRoamToV4Only(t *testing.T) { {name: "peer sends, lh over v4", themSends: true, lhOverV4: true}, {name: "peer sends, punchy", themSends: true, punchy: true}, {name: "peer sends, punchy all", themSends: true, punchy: true, punchAll: true}, + {name: "laptop sends, peer NAT", meSends: true, peerNAT: true}, + {name: "laptop sends, peer NAT, lh v4", meSends: true, peerNAT: true, lhOverV4: true}, } { t.Run(tc.name, func(t *testing.T) { t.Parallel() @@ -42,6 +44,26 @@ func TestDualStackRoamToV4Only(t *testing.T) { } } +// TestDualStackRoamRecovers holds the line TestDualStackRoamToV4Only measures: a tunnel whose remote became unreachable +// moves to the peer's other address instead of waiting out the connection manager to rehandshake, which takes 20s. +func TestDualStackRoamRecovers(t *testing.T) { + for _, tc := range []dualStackRoamCase{ + {name: "both send", meSends: true, themSends: true}, + {name: "laptop sends, peer NAT", meSends: true, peerNAT: true}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + fwd, rev := measureDualStackRoam(t, tc) + if fwd == 0 || fwd > 5*time.Second { + t.Errorf("laptop->peer recovered in %v, want within 5s", fmtRecovery(fwd)) + } + if tc.themSends && (rev == 0 || rev > 5*time.Second) { + t.Errorf("peer->laptop recovered in %v, want within 5s", fmtRecovery(rev)) + } + }) + } +} + type dualStackRoamCase struct { name string // meSends and themSends say who keeps traffic flowing after the move @@ -55,6 +77,8 @@ type dualStackRoamCase struct { simultaneous bool // punchy turns on punchy.punch everywhere, punchAll adds punchy.target_all_remotes punchy, punchAll bool + // peerNAT puts the peer behind a NAT that, after the move, only lets in traffic from addresses the peer has sent to + peerNAT bool } func fmtRecovery(d time.Duration) string { @@ -226,10 +250,21 @@ func measureDualStackRoam(t *testing.T, tc dualStackRoamCase) (time.Duration, ti done := func() bool { return (!tc.meSends || fwd.Load() != 0) && (!tc.themSends || rev.Load() != 0) } + // The peer's NAT has state for the lighthouse, which its own updates keep fresh, and for wherever it sends from here + natOpen := map[netip.AddrPort]bool{lhV4: true, lhV6: true} + isPeer := func(a netip.AddrPort) bool { return a == them.v4 || a == them.v6 } + r.RouteForAllExitFuncOrTimeout(time.Second*90, func(p *udp.Packet, _ *nebula.Control) router.ExitType { if p.To == officeV4 || p.To == officeV6Addr { return router.Drop } + if tc.peerNAT { + if isPeer(p.From) { + natOpen[p.To] = true + } else if isPeer(p.To) && !natOpen[p.From] { + return router.Drop + } + } if done() { return router.ExitNow } diff --git a/e2e/router/router.go b/e2e/router/router.go index 30055555..4e271237 100644 --- a/e2e/router/router.go +++ b/e2e/router/router.go @@ -967,7 +967,8 @@ func (r *R) getControl(p *udp.Packet) *nebula.Control { c, ok := r.inNat[toAddr] if ok { - r.outNat[outNatKey{from: c.GetUDPAddr(), to: fromAddr}] = toAddr + // Key the return path on the address the receiver will answer from, which is per family for a dual stack node + r.outNat[outNatKey{from: c.GetUDPAddrFor(fromAddr), to: fromAddr}] = toAddr return c } diff --git a/hostmap.go b/hostmap.go index 5f0838c2..49fecb5e 100644 --- a/hostmap.go +++ b/hostmap.go @@ -775,11 +775,18 @@ const unreachableProbeInterval = time.Second // remoteUnreachable is called when a send to our current remote failed because we have no route to it, typically // after moving to a network without that address family. Sends a test packet to every other remote we know for this // host: the first one that arrives roams the peer over to wherever we are now, and its reply roams us onto that path. +// +// A peer behind a NAT drops a test from an address it has never sent to, so every probe also asks the lighthouse +// about the peer. The lighthouse tells the peer where we are now, the peer punches us, and the next probe gets in. func (i *HostInfo) remoteUnreachable(ifce *Interface, dead netip.AddrPort) { if i.remotes == nil || i.ConnectionState == nil || i.GetRemote() != dead { // Nothing to probe with, or we already moved on from the remote that failed return } + if ifce.lightHouse.amLighthouse { + // A lighthouse hands out the address it reaches a client on, it must not go roaming its clients itself + return + } now := time.Now().UnixNano() next := i.nextUnreachableProbe.Load() @@ -789,12 +796,24 @@ func (i *HostInfo) remoteUnreachable(ifce *Interface, dead netip.AddrPort) { nb := make([]byte, 12) out := make([]byte, mtu) + probed := 0 i.remotes.ForEach(ifce.hostMap.GetPreferredRanges(), func(addr netip.AddrPort, _ bool) { if addr == dead { return } + probed++ ifce.sendTo(header.Test, header.TestRequest, i.ConnectionState, i, addr, []byte(""), nb, out) }) + + if probed == 0 { + // With nothing else to try, the lighthouse is our only way to learn of another remote. The peer may not have + // one, so ask on the usual requery schedule rather than every probe for as long as the tunnel sits here. + if now < i.nextLHQuery.Load() { + return + } + i.nextLHQuery.Store(now + ifce.reQueryWait.Load()) + } + ifce.lightHouse.QueryServer(i.vpnAddrs[0]) } // Bits within HostInfo.state, everything above stateEpochShift is the epoch diff --git a/inside.go b/inside.go index a26e0031..f7e225b1 100644 --- a/inside.go +++ b/inside.go @@ -215,7 +215,7 @@ func (f *Interface) sendInsideMessage(hostinfo *HostInfo, pkt tio.Packet, nb []b return nil } - sendBatch.Commit(toSend, relayHostInfo.GetRemote(), relayHostInfo.localIndexId) + sendBatch.Commit(toSend, relayHostInfo.GetRemote()) return nil }) if err != nil { @@ -233,7 +233,7 @@ func (f *Interface) sendInsideMessage(hostinfo *HostInfo, pkt tio.Packet, nb []b return nil } - sendBatch.Commit(out, remote, hostinfo.localIndexId) + sendBatch.Commit(out, remote) return nil }) if err != nil { @@ -612,7 +612,8 @@ func (f *Interface) sendNoMetrics(t header.MessageType, st header.MessageSubType "udpAddr", hr, ) } - if udp.IsUnreachable(err) { + // A closing tunnel is about to be torn down, there is no point looking for another path for it + if t != header.CloseTunnel && udp.IsUnreachable(err) { hostinfo.remoteUnreachable(f, hr) } } diff --git a/interface.go b/interface.go index 6ed68ec5..9e7fcfdc 100644 --- a/interface.go +++ b/interface.go @@ -1,6 +1,7 @@ package nebula import ( + "bytes" "context" "crypto/fips140" "errors" @@ -507,24 +508,40 @@ func (f *Interface) flushSendBatch(sb *batch.SendBatch, q int) { f.metricTxDropped.Inc(int64(dropped)) } - // Packets are tagged with the local index of the tunnel they belong to. A tunnel's packets tend to fail together, - // so skip straight past a repeat of the one before. + // A tunnel's packets tend to fail together, so skip straight past a repeat of the one before. The first 8 bytes + // of a header say whose packet it is, the counter after them changes with every packet. var last batch.Failure for i, fail := range sb.Failures() { - if i > 0 && fail.Tag == last.Tag && fail.Dst == last.Dst { + if i > 0 && fail.Dst == last.Dst && bytes.Equal(fail.Head[:8], last.Head[:8]) { continue } last = fail - if !udp.IsUnreachable(fail.Err) { - continue - } - if hostinfo := f.hostMap.QueryIndex(fail.Tag); hostinfo != nil { - hostinfo.remoteUnreachable(f, fail.Dst) + if udp.IsUnreachable(fail.Err) { + f.sendUnreachable(fail.Head[:], fail.Dst) } } } +// sendUnreachable finds the tunnel a packet we had no route for was sent on, from the start of the packet, and has it +// look for another path to its peer. +func (f *Interface) sendUnreachable(head []byte, dst netip.AddrPort) { + var h header.H + if h.Parse(head) != nil { + return + } + + if h.Type == header.Message && h.Subtype == header.MessageRelay { + // TODO: a relayed packet's header carries the relay session's index, not the relay tunnel's, so it does not + // lead us to the tunnel whose remote failed. The relay tunnel's own traffic finds out soon enough. + return + } + + if hostinfo := f.hostMap.QueryReverseIndex(h.RemoteIndex); hostinfo != nil { + hostinfo.remoteUnreachable(f, dst) + } +} + func (f *Interface) RegisterConfigChangeCallbacks(c *config.C) { c.RegisterReloadCallback(f.reloadFirewall) for _, reload := range []func(*config.C){ diff --git a/overlay/batch/tx_batch.go b/overlay/batch/tx_batch.go index 23dfca13..14a58ba4 100644 --- a/overlay/batch/tx_batch.go +++ b/overlay/batch/tx_batch.go @@ -4,16 +4,20 @@ import "net/netip" const SendBatchCap = 128 +// FailureHeadLen is how much of a refused packet a Failure keeps, nebula's header, which says whose packet it was +const FailureHeadLen = 16 + // batchWriter is the minimal subset of udp.Conn needed by SendBatch to flush. type batchWriter interface { WriteBatch(bufs [][]byte, addrs []netip.AddrPort, failed func(i int, err error)) (int, error) } -// Failure is a packet the underlay would not send, with the tag it was committed under +// Failure is a packet the underlay would not send type Failure struct { - Tag uint32 - Dst netip.AddrPort - Err error + // Head is the start of the packet, zero padded if the packet was shorter + Head [FailureHeadLen]byte + Dst netip.AddrPort + Err error } // SendBatch accumulates encrypted UDP packets and flushes them via WriteBatch. @@ -23,7 +27,6 @@ type SendBatch struct { out batchWriter bufs [][]byte dsts []netip.AddrPort - tags []uint32 arena *Arena failures []Failure @@ -37,7 +40,6 @@ func NewSendBatch(out batchWriter, batchCap, arenaSize int) *SendBatch { out: out, bufs: make([][]byte, 0, batchCap), dsts: make([]netip.AddrPort, 0, batchCap), - tags: make([]uint32, 0, batchCap), arena: NewArena(arenaSize), } b.recordFailure = b.failed @@ -45,7 +47,9 @@ func NewSendBatch(out batchWriter, batchCap, arenaSize int) *SendBatch { } func (b *SendBatch) failed(i int, err error) { - b.failures = append(b.failures, Failure{Tag: b.tags[i], Dst: b.dsts[i], Err: err}) + f := Failure{Dst: b.dsts[i], Err: err} + copy(f.Head[:], b.bufs[i]) + b.failures = append(b.failures, f) } func (b *SendBatch) Reserve(sz int) []byte { @@ -57,12 +61,9 @@ func (b *SendBatch) Reserve(sz int) []byte { // bounding how long the first packet of a large read batch waits. func (b *SendBatch) Len() int { return len(b.bufs) } -// Commit queues pkt for dst. tag comes back in Failures if the underlay refuses the packet, so the caller can tell -// whose packet it was. -func (b *SendBatch) Commit(pkt []byte, dst netip.AddrPort, tag uint32) { +func (b *SendBatch) Commit(pkt []byte, dst netip.AddrPort) { b.bufs = append(b.bufs, pkt) b.dsts = append(b.dsts, dst) - b.tags = append(b.tags, tag) } // Flush writes every queued packet and reports how many actually went out. A short count means some destinations @@ -79,7 +80,6 @@ func (b *SendBatch) Flush() (int, error) { clear(b.bufs) b.bufs = b.bufs[:0] b.dsts = b.dsts[:0] - b.tags = b.tags[:0] b.arena.Reset() return written, err } diff --git a/overlay/batch/tx_batch_test.go b/overlay/batch/tx_batch_test.go index 7e5aa024..5f1d0a69 100644 --- a/overlay/batch/tx_batch_test.go +++ b/overlay/batch/tx_batch_test.go @@ -36,7 +36,7 @@ func TestSendBatchReserveCommitFlush(t *testing.T) { t.Fatalf("slot %d: cap=%d want 32", i, cap(slot)) } pkt := append(slot[:0], byte(i), byte(i+1), byte(i+2)) - b.Commit(pkt, ap, 0) + b.Commit(pkt, ap) } if _, err := b.Flush(); err != nil { t.Fatalf("Flush: %v", err) @@ -77,7 +77,7 @@ func TestSendBatchSlotsDoNotOverlap(t *testing.T) { for i := range 3 { s := b.Reserve(8) pkt := append(s[:0], byte(0xA0+i), byte(0xB0+i)) - b.Commit(pkt, ap, 0) + b.Commit(pkt, ap) } if _, err := b.Flush(); err != nil { t.Fatalf("Flush: %v", err) @@ -98,11 +98,11 @@ func TestSendBatchGrowPreservesCommitted(t *testing.T) { s1 := b.Reserve(4) pkt1 := append(s1[:0], 0x11, 0x22, 0x33, 0x44) - b.Commit(pkt1, ap, 0) + b.Commit(pkt1, ap) s2 := b.Reserve(8) // exceeds remaining cap, triggers grow pkt2 := append(s2[:0], 0xA, 0xB, 0xC, 0xD, 0xE) - b.Commit(pkt2, ap, 0) + b.Commit(pkt2, ap) // pkt1 must still be intact even though backing reallocated. if pkt1[0] != 0x11 || pkt1[3] != 0x44 { @@ -140,11 +140,24 @@ func (w *failingBatchWriter) WriteBatch(bufs [][]byte, _ []netip.AddrPort, faile return written, nil } -func TestSendBatchReportsFailedTags(t *testing.T) { +func TestSendBatchReportsFailures(t *testing.T) { errNoRoute := errors.New("no route") fw := &failingBatchWriter{refuse: map[byte]error{2: errNoRoute, 3: errNoRoute}} + b := NewSendBatch(fw, 4, 128) - b := NewSendBatch(fw, 4, 64) + // pkt is n bytes starting with the refuse key, each later byte its offset plus 0x10 so the head is recognizable + pkt := func(key byte, n int) []byte { + p := b.Reserve(n)[:n] + p[0] = key + for i := 1; i < n; i++ { + p[i] = byte(0x10 + i) + } + return p + } + head := func(p []byte) (h [FailureHeadLen]byte) { + copy(h[:], p) + return h + } dsts := []netip.AddrPort{ netip.MustParseAddrPort("10.0.0.1:4242"), @@ -152,10 +165,12 @@ func TestSendBatchReportsFailedTags(t *testing.T) { netip.MustParseAddrPort("[fd00::3]:4242"), netip.MustParseAddrPort("10.0.0.4:4242"), } - for i, dst := range dsts { - pkt := append(b.Reserve(8)[:0], byte(i+1)) - b.Commit(pkt, dst, uint32(100+i)) - } + long := pkt(2, FailureHeadLen+8) // only its head is kept + short := pkt(3, 4) // zero padded + b.Commit(pkt(1, 20), dsts[0]) + b.Commit(long, dsts[1]) + b.Commit(short, dsts[2]) + b.Commit(pkt(1, 20), dsts[3]) written, err := b.Flush() if err != nil { @@ -164,22 +179,23 @@ func TestSendBatchReportsFailedTags(t *testing.T) { if written != 2 { t.Errorf("written = %d, want 2", written) } - want := []Failure{{101, dsts[1], errNoRoute}, {102, dsts[2], errNoRoute}} + want := []Failure{{head(long), dsts[1], errNoRoute}, {head(short), dsts[2], errNoRoute}} if got := b.Failures(); !slices.Equal(got, want) { t.Errorf("failures = %v, want %v", got, want) } - // Tags do not leak into the next flush - b.Commit(append(b.Reserve(8)[:0], 2), dsts[0], 7) + // Failures do not leak into the next flush + again := pkt(2, 20) + b.Commit(again, dsts[0]) if _, err := b.Flush(); err != nil { t.Fatalf("Flush: %v", err) } - if want := []Failure{{7, dsts[0], errNoRoute}}; !slices.Equal(b.Failures(), want) { + if want := []Failure{{head(again), dsts[0], errNoRoute}}; !slices.Equal(b.Failures(), want) { t.Errorf("second flush failures = %v, want %v", b.Failures(), want) } // A clean flush forgets the last one's failures - b.Commit(append(b.Reserve(8)[:0], 1), dsts[0], 8) + b.Commit(pkt(1, 20), dsts[0]) if _, err := b.Flush(); err != nil { t.Fatalf("Flush: %v", err) } diff --git a/punchy_test.go b/punchy_test.go index 47f09af0..3096f54d 100644 --- a/punchy_test.go +++ b/punchy_test.go @@ -8,6 +8,7 @@ import ( "time" "github.com/slackhq/nebula/config" + "github.com/slackhq/nebula/header" "github.com/slackhq/nebula/test" "github.com/slackhq/nebula/udp" "github.com/stretchr/testify/assert" @@ -248,18 +249,37 @@ func TestPunchyReloadRestartRequired(t *testing.T) { // errNoRoute is the unreachable error every platform classifies the same way var errNoRoute = udp.ErrInvalidIPv6RemoteForSocket -// recordingConn records where every write went and fails the ones to an address in fail +// recordingConn records where every write went and what kind of message it was, and fails the ones to an address in +// fail type recordingConn struct { udp.NoopConn fail map[netip.AddrPort]error wrote []netip.AddrPort + // types is the message type of each write, a punch has none + types []header.MessageType } -func (c *recordingConn) WriteTo(_ []byte, addr netip.AddrPort) error { +func (c *recordingConn) WriteTo(b []byte, addr netip.AddrPort) error { c.wrote = append(c.wrote, addr) + var h header.H + if h.Parse(b) == nil { + c.types = append(c.types, h.Type) + } return c.fail[addr] } +func (c *recordingConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort, failed func(int, error)) (int, error) { + written := 0 + for i := range bufs { + if err := c.WriteTo(bufs[i], addrs[i]); err != nil { + failed(i, err) + continue + } + written++ + } + return written, nil +} + type noLighthouses struct{} func (noLighthouses) IsAnyLighthouseAddr([]netip.Addr) bool { return false } diff --git a/udp/udp_tester.go b/udp/udp_tester.go index 2a7340cf..c9161c89 100644 --- a/udp/udp_tester.go +++ b/udp/udp_tester.go @@ -129,6 +129,14 @@ func (u *TesterConn) SetDualStackAddrs(v4, v6 netip.AddrPort) { u.addrs.Store(a) } +// GetAddrFor returns the underlay address a send to dst leaves from, GetAddr when there is no route to dst +func (u *TesterConn) GetAddrFor(dst netip.AddrPort) netip.AddrPort { + if src, err := u.sourceFor(dst); err == nil { + return src + } + return u.GetAddr() +} + // sourceFor returns the address a send to dst leaves from func (u *TesterConn) sourceFor(dst netip.AddrPort) (netip.AddrPort, error) { a := u.addrs.Load()