From 7902ce674e3df74c28316447fd770b0683456ad3 Mon Sep 17 00:00:00 2001 From: Nate Brown Date: Thu, 23 Jul 2026 09:26:24 -0500 Subject: [PATCH] Rebind for MacOS (#1816) Co-authored-by: Jack Doan --- connection_manager_test.go | 1 + control.go | 10 +- control_tester.go | 14 +- e2e/rebind_test.go | 225 ++++++++++++++++++++++++++++++++ e2e/router/router.go | 156 ++++++++++++++++++---- examples/config.yml | 8 ++ lighthouse.go | 10 +- main.go | 3 + udp/netchange.go | 61 +++++++++ udp/netchange_darwin.go | 164 +++++++++++++++++++++++ udp/netchange_darwin_test.go | 244 +++++++++++++++++++++++++++++++++++ udp/netchange_generic.go | 22 ++++ udp/netchange_test.go | 39 ++++++ udp/udp_darwin.go | 9 +- udp/udp_tester.go | 26 +++- 15 files changed, 951 insertions(+), 41 deletions(-) create mode 100644 e2e/rebind_test.go create mode 100644 udp/netchange.go create mode 100644 udp/netchange_darwin.go create mode 100644 udp/netchange_darwin_test.go create mode 100644 udp/netchange_generic.go create mode 100644 udp/netchange_test.go diff --git a/connection_manager_test.go b/connection_manager_test.go index e167e5f2..25637c25 100644 --- a/connection_manager_test.go +++ b/connection_manager_test.go @@ -25,6 +25,7 @@ func newTestLighthouse() *LightHouse { lighthouses := []netip.Addr{} staticList := map[netip.Addr]struct{}{} + lh.localAddrsFn = func(*LocalAllowList) []netip.Addr { return nil } lh.lighthouses.Store(&lighthouses) lh.staticList.Store(&staticList) diff --git a/control.go b/control.go index a79ebbfa..7df5a09e 100644 --- a/control.go +++ b/control.go @@ -53,6 +53,7 @@ type Control struct { statsStart func() dnsStart func() lighthouseStart func() + networkChangeStart func(rebind func()) connectionManagerStart func(context.Context) } @@ -104,6 +105,9 @@ func (c *Control) Start() error { if c.dnsStart != nil { go c.dnsStart() } + if c.networkChangeStart != nil { + go c.networkChangeStart(c.RebindUDPServer) + } if c.connectionManagerStart != nil { go c.connectionManagerStart(c.ctx) } @@ -198,7 +202,11 @@ func (c *Control) RebindUDPServer() { return } - _ = c.f.outside.Rebind() + // A failure here means we are likely still pinned to the interface we came up on, so the rest of this is + // unlikely to help. Say so instead of silently carrying on as if we rebound. + if err := c.f.outside.Rebind(); err != nil { + c.l.Error("Failed to rebind udp socket", "error", err) + } // Trigger a lighthouse update, useful for mobile clients that should have an update interval of 0 c.f.lightHouse.SendUpdate() diff --git a/control_tester.go b/control_tester.go index 422d86ec..546b9e87 100644 --- a/control_tester.go +++ b/control_tester.go @@ -108,7 +108,19 @@ func (c *Control) GetVpnAddrs() []netip.Addr { } func (c *Control) GetUDPAddr() netip.AddrPort { - return c.f.outside.(*udp.TesterConn).Addr + return c.f.outside.(*udp.TesterConn).GetAddr() +} + +// SetUDPAddr moves this node to a new underlay address, standing in for a laptop waking up on a different +// network. Register the new address with the router as well or nothing will route back. +func (c *Control) SetUDPAddr(addr netip.AddrPort) { + c.f.outside.(*udp.TesterConn).SetAddr(addr) +} + +// SetLocalAddrsFn replaces underlay address discovery so a test can advertise its simulated address instead of +// whatever this machine's NICs happen to be. Call it before Start, SendUpdate reads it from the update worker. +func (c *Control) SetLocalAddrsFn(fn func(*LocalAllowList) []netip.Addr) { + c.f.lightHouse.localAddrsFn = fn } func (c *Control) KillPendingTunnel(vpnIp netip.Addr) bool { diff --git a/e2e/rebind_test.go b/e2e/rebind_test.go new file mode 100644 index 00000000..2547f739 --- /dev/null +++ b/e2e/rebind_test.go @@ -0,0 +1,225 @@ +//go:build e2e_testing +// +build e2e_testing + +package e2e + +import ( + "net/netip" + "testing" + "time" + + "github.com/slackhq/nebula" + "github.com/slackhq/nebula/cert" + "github.com/slackhq/nebula/cert_test" + "github.com/slackhq/nebula/e2e/router" + "github.com/slackhq/nebula/header" + "github.com/slackhq/nebula/udp" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// reportedAddrs is what the lighthouse would hand a peer asking where vpnAddr is. +func reportedAddrs(t *testing.T, lh *nebula.Control, vpnAddr netip.Addr) []netip.AddrPort { + t.Helper() + cm := lh.QueryLighthouse(vpnAddr) + if cm == nil { + return nil + } + var out []netip.AddrPort + for _, c := range *cm { + out = append(out, c.Reported...) + out = append(out, c.Learned...) + } + return out +} + +// waitForLighthouseMsg routes until a lighthouse message lands on lh, or gives up. Reports whether one arrived. +func waitForLighthouseMsg(t *testing.T, r *router.R, lh *nebula.Control, wait time.Duration) bool { + t.Helper() + h := &header.H{} + return r.RouteForAllExitFuncOrTimeout(wait, func(p *udp.Packet, c *nebula.Control) router.ExitType { + if c != lh { + return router.KeepRouting + } + // Punches are a single byte and never parse, they are just not what we are after + if err := h.Parse(p.Data); err != nil { + return router.KeepRouting + } + if h.Type == header.LightHouse { + return router.RouteAndExit + } + return router.KeepRouting + }) +} + +// A laptop that changes networks has to tell the lighthouse promptly, otherwise the lighthouse keeps handing peers +// the old address and their punches land nowhere. On a long lighthouse interval the only thing that closes that +// window is the rebind, which on darwin the network change monitor drives. The e2e build compiles the monitor out, +// so we call RebindUDPServer directly, which is the same thing the monitor does. +func TestRebindSendsLighthouseUpdate(t *testing.T) { + t.Parallel() + ca, _, caKey, _ := cert_test.NewTestCaCert(cert.Version2, cert.Curve_CURVE25519, time.Now(), time.Now().Add(10*time.Minute), nil, nil, []string{}) + + lhControl, lhVpnIpNet, lhUdpAddr, _ := newSimpleServer(cert.Version2, ca, caKey, "lh", "10.128.0.1/24", m{ + "lighthouse": m{"am_lighthouse": true}, + }) + + // 600s interval, so nothing scheduled can send an update during this test. A rebind is the only thing that can. + myControl, _, _, _ := newSimpleServer(cert.Version2, ca, caKey, "me", "10.128.0.2/24", m{ + "lighthouse": m{ + "hosts": []any{lhVpnIpNet[0].Addr().String()}, + "interval": 600, + }, + "static_host_map": m{ + lhVpnIpNet[0].Addr().String(): []any{lhUdpAddr.String()}, + }, + }) + + r := router.NewR(t, lhControl, myControl) + defer r.RenderFlow() + + lhControl.Start() + myControl.Start() + + // Let the startup registration finish, then clear everything it left behind + require.True(t, waitForLighthouseMsg(t, r, lhControl, time.Second*5), "expected an initial registration") + r.RouteFor(time.Millisecond * 400) + + // Nothing should be talking to the lighthouse on its own now + require.False(t, waitForLighthouseMsg(t, r, lhControl, time.Millisecond*200), + "nothing should reach the lighthouse before the rebind") + + myControl.RebindUDPServer() + + assert.True(t, waitForLighthouseMsg(t, r, lhControl, time.Second*5), + "a rebind should push an update to the lighthouse rather than waiting out the interval") + + lhControl.Stop() + myControl.Stop() +} + +// The other half of a rebind: every live tunnel requeries the lighthouse on its next send. That query is what makes +// the lighthouse tell the peer to punch toward our new address, which is the part that actually revives a tunnel +// whose remote NAT state died while we were on a different network. +func TestRebindRequeriesPeersOnNextSend(t *testing.T) { + t.Parallel() + ca, _, caKey, _ := cert_test.NewTestCaCert(cert.Version2, cert.Curve_CURVE25519, time.Now(), time.Now().Add(10*time.Minute), nil, nil, []string{}) + + lhControl, lhVpnIpNet, lhUdpAddr, _ := newSimpleServer(cert.Version2, ca, caKey, "lh", "10.128.0.1/24", m{ + "lighthouse": m{"am_lighthouse": true}, + }) + + lhCfg := m{ + "lighthouse": m{ + "hosts": []any{lhVpnIpNet[0].Addr().String()}, + "interval": 600, + // Without this the peers advertise this machine's real addresses and then try to punch at them, + // which the router has no route for. + "local_allow_list": m{ + "10.0.0.0/24": true, + "::/0": false, + }, + }, + "static_host_map": m{ + lhVpnIpNet[0].Addr().String(): []any{lhUdpAddr.String()}, + }, + } + + myControl, myVpnIpNet, myUdpAddr, _ := newSimpleServer(cert.Version2, ca, caKey, "me", "10.128.0.2/24", lhCfg) + theirControl, theirVpnIpNet, theirUdpAddr, _ := newSimpleServer(cert.Version2, ca, caKey, "them", "10.128.0.3/24", lhCfg) + + r := router.NewR(t, lhControl, myControl, theirControl) + defer r.RenderFlow() + + lhControl.Start() + myControl.Start() + theirControl.Start() + r.RouteFor(time.Millisecond * 500) + + // Point the peers at each other directly, this test is about the rebind and not about lighthouse discovery + myControl.InjectLightHouseAddr(theirVpnIpNet[0].Addr(), theirUdpAddr) + theirControl.InjectLightHouseAddr(myVpnIpNet[0].Addr(), myUdpAddr) + + myControl.InjectTunPacket(BuildTunUDPPacket(theirVpnIpNet[0].Addr(), 80, myVpnIpNet[0].Addr(), 80, []byte("initial"))) + r.RouteFor(time.Second) + require.NotNil(t, myControl.GetHostInfoByVpnAddr(theirVpnIpNet[0].Addr(), false), "expected a tunnel to them") + r.RouteFor(time.Millisecond * 300) + + // Assert on what the peer sees rather than on lighthouse traffic. A query for them makes the lighthouse send + // them a punch notification, which is the whole point. Our own update to the lighthouse sends them nothing, + // so this cannot be satisfied by the update the rebind itself pushes. + myControl.InjectTunPacket(BuildTunUDPPacket(theirVpnIpNet[0].Addr(), 80, myVpnIpNet[0].Addr(), 80, []byte("quiet"))) + require.False(t, waitForLighthouseMsg(t, r, theirControl, time.Millisecond*300), + "an ordinary send should not requery the lighthouse") + + myControl.RebindUDPServer() + r.RouteFor(time.Millisecond * 300) // let the update the rebind itself sends pass by + + myControl.InjectTunPacket(BuildTunUDPPacket(theirVpnIpNet[0].Addr(), 80, myVpnIpNet[0].Addr(), 80, []byte("after rebind"))) + assert.True(t, waitForLighthouseMsg(t, r, theirControl, time.Second*5), + "the first send after a rebind should requery the lighthouse, which then tells the peer to punch at us") + + lhControl.Stop() + myControl.Stop() + theirControl.Stop() +} + +// The scenario this whole thing exists for: a laptop sleeps at the office and wakes up at home on a new address. +// Until it tells the lighthouse, the lighthouse keeps handing peers the office address, so their punches land +// nowhere and the tunnel stays dead. On a long interval the rebind is the only thing that closes that window. +func TestRebindAdvertisesNewAddressAfterMove(t *testing.T) { + t.Parallel() + ca, _, caKey, _ := cert_test.NewTestCaCert(cert.Version2, cert.Curve_CURVE25519, time.Now(), time.Now().Add(10*time.Minute), nil, nil, []string{}) + + lhControl, lhVpnIpNet, lhUdpAddr, _ := newSimpleServer(cert.Version2, ca, caKey, "lh", "10.128.0.1/24", m{ + "lighthouse": m{"am_lighthouse": true}, + }) + + myControl, myVpnIpNet, myUdpAddr, _ := newSimpleServer(cert.Version2, ca, caKey, "me", "10.128.0.2/24", m{ + "lighthouse": m{ + "hosts": []any{lhVpnIpNet[0].Addr().String()}, + "interval": 600, + }, + "static_host_map": m{ + lhVpnIpNet[0].Addr().String(): []any{lhUdpAddr.String()}, + }, + }) + + // Advertise wherever we currently are rather than this machine's real NICs, read fresh each time so a move + // is picked up. + myControl.SetLocalAddrsFn(func(*nebula.LocalAllowList) []netip.Addr { + return []netip.Addr{myControl.GetUDPAddr().Addr()} + }) + + r := router.NewR(t, lhControl, myControl) + defer r.RenderFlow() + + lhControl.Start() + myControl.Start() + + require.True(t, waitForLighthouseMsg(t, r, lhControl, time.Second*5), "expected an initial registration") + r.RouteFor(time.Millisecond * 400) + + require.Contains(t, reportedAddrs(t, lhControl, myVpnIpNet[0].Addr()), myUdpAddr, + "the lighthouse should know the address we started on") + + // Wake up somewhere else + newAddr := netip.MustParseAddrPort("10.0.0.99:4242") + myControl.SetUDPAddr(newAddr) + r.AddRoute(newAddr.Addr(), newAddr.Port(), myControl) + + // Nothing has told the lighthouse, and with interval 600 nothing scheduled will + r.RouteFor(time.Millisecond * 400) + require.NotContains(t, reportedAddrs(t, lhControl, myVpnIpNet[0].Addr()), newAddr, + "the lighthouse should still be handing out the old address before the rebind") + + myControl.RebindUDPServer() + require.True(t, waitForLighthouseMsg(t, r, lhControl, time.Second*5), "expected an update after the rebind") + r.RouteFor(time.Millisecond * 400) + + assert.Contains(t, reportedAddrs(t, lhControl, myVpnIpNet[0].Addr()), newAddr, + "after the rebind the lighthouse should hand peers our new address") + + lhControl.Stop() + myControl.Stop() +} diff --git a/e2e/router/router.go b/e2e/router/router.go index 72012073..9f040e4a 100644 --- a/e2e/router/router.go +++ b/e2e/router/router.go @@ -114,6 +114,28 @@ type packet struct { packet *udp.Packet tun bool // a packet pulled off a tun device rx bool // the packet was received by a udp device + + // h is the nebula header, parsed once when the packet is recorded. parseErr says why there isn't one, which + // the flow log reports rather than hiding. Punchy sends a single byte, so an unparseable packet is normal. + h header.H + parseErr error +} + +// fromAddr and toAddr are the addresses this packet actually travelled between. Reading them off the control +// instead would misreport the whole history once a test moves a node. Tun packets are synthesized without +// addresses, so they fall back to the control. +func (p *packet) fromAddr() netip.AddrPort { + if p.tun || !p.packet.From.IsValid() { + return p.from.GetUDPAddr() + } + return p.packet.From +} + +func (p *packet) toAddr() netip.AddrPort { + if p.tun || !p.packet.To.IsValid() { + return p.to.GetUDPAddr() + } + return p.packet.To } func (p *packet) WasReceived() { @@ -249,7 +271,7 @@ func (r *R) renderFlow() { continue } - addr := e.packet.from.GetUDPAddr() + addr := e.packet.fromAddr() if _, ok := participants[addr]; ok { continue } @@ -268,7 +290,6 @@ func (r *R) renderFlow() { } // Print packets - h := &header.H{} for _, e := range r.flow { if e.packet == nil { //fmt.Fprintf(f, " note over %s: %s\n", strings.Join(participantsVals, ", "), e.note) @@ -280,21 +301,22 @@ func (r *R) renderFlow() { fmt.Fprintln(f, r.formatUdpPacket(p)) } else { - if err := h.Parse(p.packet.Data); err != nil { - panic(err) - } - line := "--x" if p.rx { line = "->>" } - fmt.Fprintf(f, - " %s%s%s: %s(%s), index %v, counter: %v\n", - normalizeName(p.from.GetUDPAddr().String()), + detail := fmt.Sprintf("%s(%s), index %v, counter: %v", + p.h.TypeName(), p.h.SubTypeName(), p.h.RemoteIndex, p.h.MessageCounter) + if p.parseErr != nil { + detail = fmt.Sprintf("unparsed, %v (%d bytes)", p.parseErr, len(p.packet.Data)) + } + + fmt.Fprintf(f, " %s%s%s: %s\n", + normalizeName(p.fromAddr().String()), line, - normalizeName(p.to.GetUDPAddr().String()), - h.TypeName(), h.SubTypeName(), h.RemoteIndex, h.MessageCounter, + normalizeName(p.toAddr().String()), + detail, ) } } @@ -408,29 +430,34 @@ func (r *R) unlockedInjectFlow(from, to *nebula.Control, p *udp.Packet, tun bool r.renderHostmaps(fmt.Sprintf("Packet %v", len(r.flow))) - if len(r.ignoreFlows) > 0 { - var h header.H - err := h.Parse(p.Data) - if err != nil { - panic(err) - } + var h header.H + var parseErr error + if !tun { + parseErr = h.Parse(p.Data) + } - for _, i := range r.ignoreFlows { - if !tun { - if i.messageType == h.Type && i.subType == h.Subtype { - return nil - } - } else if i.tun.HasValue && i.tun.IsTrue { + // Decide before copying, the copy comes from a freelist and an ignored packet would never be released + for _, i := range r.ignoreFlows { + if tun { + if i.tun.HasValue && i.tun.IsTrue { return nil } + continue + } + + // A packet we could not parse has no type to match against, so no rule can ignore it + if parseErr == nil && i.messageType == h.Type && i.subType == h.Subtype { + return nil } } fp := &packet{ - from: from, - to: to, - packet: p.Copy(), - tun: tun, + from: from, + to: to, + packet: p.Copy(), + tun: tun, + h: h, + parseErr: parseErr, } r.flow = append(r.flow, flowEntry{packet: fp}) @@ -690,6 +717,81 @@ func (r *R) RouteUntilAfterMsgType(sender *nebula.Control, msgType header.Messag }) } +// RouteFor routes everything that shows up for the given duration and then returns. Use it to let a test settle +// deterministically rather than sleeping and hoping: a single FlushAll races a completing handshake, which queues +// more packets right behind it. +func (r *R) RouteFor(d time.Duration) { + r.RouteForAllExitFuncOrTimeout(d, func(*udp.Packet, *nebula.Control) ExitType { + return KeepRouting + }) +} + +// RouteForAllExitFuncOrTimeout is RouteForAllExitFunc with a deadline, reporting whether whatDo asked to exit +// before time ran out. The unbounded version blocks forever on a quiet network, so this is what a test needs to +// assert that something does NOT happen, or to route for a fixed settling period. +func (r *R) RouteForAllExitFuncOrTimeout(timeout time.Duration, whatDo ExitFunc) bool { + sc := make([]reflect.SelectCase, 0, len(r.controls)+1) + cm := make([]*nebula.Control, 0, len(r.controls)) + + for _, c := range r.controls { + sc = append(sc, reflect.SelectCase{ + Dir: reflect.SelectRecv, + Chan: reflect.ValueOf(c.GetUDPTxChan()), + Send: reflect.Value{}, + }) + cm = append(cm, c) + } + + timer := time.NewTimer(timeout) + defer timer.Stop() + sc = append(sc, reflect.SelectCase{ + Dir: reflect.SelectRecv, + Chan: reflect.ValueOf(timer.C), + Send: reflect.Value{}, + }) + + for { + x, rx, _ := reflect.Select(sc) + if x == len(cm) { + return false + } + + r.Lock() + p := rx.Interface().(*udp.Packet) + receiver := r.getControl(cm[x].GetUDPAddr(), p.To, p) + if receiver == nil { + r.Unlock() + panic("Can't RouteForAllExitFuncOrTimeout for host: " + p.To.String()) + } + + e := whatDo(p, receiver) + switch e { + case ExitNow: + r.Unlock() + p.Release() + return true + + case RouteAndExit: + fp := r.unlockedInjectFlow(cm[x], receiver, p, false) + receiver.InjectUDPPacket(p) + fp.WasReceived() + r.Unlock() + p.Release() + return true + + case KeepRouting: + fp := r.unlockedInjectFlow(cm[x], receiver, p, false) + receiver.InjectUDPPacket(p) + fp.WasReceived() + + default: + panic(fmt.Sprintf("Unknown exitFunc return: %v", e)) + } + r.Unlock() + p.Release() + } +} + func (r *R) RouteForAllUntilAfterMsgTypeTo(receiver *nebula.Control, msgType header.MessageType, subType header.MessageSubType) { h := &header.H{} r.RouteForAllExitFunc(func(p *udp.Packet, r *nebula.Control) ExitType { diff --git a/examples/config.yml b/examples/config.yml index 4f7fd1e7..d5409bce 100644 --- a/examples/config.yml +++ b/examples/config.yml @@ -146,6 +146,14 @@ listen: # Default true; set to false to leave WDF in charge of inbound decisions on the listener port. Not reloadable. #windows_bypass_wdf: true + # On macOS only + # macOS scopes the udp socket to the interface it was created on, so moving between networks (wifi to wired, + # office to home) leaves Nebula sending out an interface that no longer has a route. When true, Nebula watches + # the routing socket and rebinds the listener once the change settles. + # iOS does not use this, the host app drives the same rebind itself. + # Default true. Not reloadable. + #rebind_on_network_change: true + # By default, Nebula replies to packets it has no tunnel for with a "recv_error" packet. This packet helps speed up reconnection # in the case that Nebula on either side did not shut down cleanly. This response can be abused as a way to discover if Nebula is running # on a host though. This option lets you configure if you want to send "recv_error" packets always, never, or only to private network remotes. diff --git a/lighthouse.go b/lighthouse.go index 3df74c39..9cece233 100644 --- a/lighthouse.go +++ b/lighthouse.go @@ -36,6 +36,10 @@ type LightHouse struct { myVpnNetworksTable *bart.Lite punchy *Punchy + // localAddrsFn enumerates the underlay addresses we advertise. It is a field so tests can supply simulated + // addresses rather than whatever this machine's NICs happen to be. Set it before Start. + localAddrsFn func(*LocalAllowList) []netip.Addr + // Local cache of answers from light houses // map of vpn addr to answers addrMap map[netip.Addr]*RemoteList @@ -107,6 +111,10 @@ func NewLightHouseFromConfig(ctx context.Context, l *slog.Logger, c *config.C, c queryChan: make(chan netip.Addr, c.GetUint32("handshakes.query_buffer", 64)), l: l, } + h.localAddrsFn = func(al *LocalAllowList) []netip.Addr { + return localAddrs(h.l, al) + } + lighthouses := make([]netip.Addr, 0) h.lighthouses.Store(&lighthouses) staticList := make(map[netip.Addr]struct{}) @@ -918,7 +926,7 @@ func (lh *LightHouse) SendUpdate() { } lal := lh.GetLocalAllowList() - for _, e := range localAddrs(lh.l, lal) { + for _, e := range lh.localAddrsFn(lal) { if lh.myVpnNetworksTable.Contains(e) { continue } diff --git a/main.go b/main.go index d62d8dd0..2ef2031b 100644 --- a/main.go +++ b/main.go @@ -268,6 +268,8 @@ func Main(c *config.C, configTest bool, buildVersion string, l *slog.Logger, dev attachCommands(l, c, ssh, ifce) + networkChanges := udp.NewNetworkChangeMonitor(ctx, l, c) + return &Control{ state: StateReady, f: ifce, @@ -278,6 +280,7 @@ func Main(c *config.C, configTest bool, buildVersion string, l *slog.Logger, dev statsStart: stats.Start, dnsStart: ds.Start, lighthouseStart: lightHouse.StartUpdateWorker, + networkChangeStart: networkChanges.Start, connectionManagerStart: connManager.Start, }, nil } diff --git a/udp/netchange.go b/udp/netchange.go new file mode 100644 index 00000000..e7b21e6b --- /dev/null +++ b/udp/netchange.go @@ -0,0 +1,61 @@ +package udp + +import ( + "context" + "log/slog" + + "github.com/slackhq/nebula/config" +) + +// NetworkChangeMonitor rebinds the udp listener when the local network moves out from under it. +// +// Detection lives here in the udp package, next to the socket it concerns and the platform matrix that already knows +// which sockets go stale. What to do about a change — updating the lighthouse, requerying tunnels — is not the udp +// package's business, so Start takes the reaction as a plain function. Passing it at Start rather than holding it +// keeps this package from referencing whatever owns the rebind. +// +// On platforms whose sockets do not go stale, watchNetworkChanges hands back a nil channel and Start returns. +type NetworkChangeMonitor struct { + l *slog.Logger + ctx context.Context + enabled bool +} + +// NewNetworkChangeMonitor builds a monitor for local network changes. The returned monitor is always usable: Start +// is safe to call unconditionally, it no-ops when disabled or on a platform that does not need it. +func NewNetworkChangeMonitor(ctx context.Context, l *slog.Logger, c *config.C) *NetworkChangeMonitor { + return &NetworkChangeMonitor{ + l: l, + ctx: ctx, + enabled: c.GetBool("listen.rebind_on_network_change", true), + } +} + +// Start watches for network changes until the context is cancelled, calling rebind once per settled change. It +// blocks, so callers run it in a goroutine, and it no-ops when disabled, unsupported, or with nothing to rebind. +func (m *NetworkChangeMonitor) Start(rebind func()) { + if !m.enabled || rebind == nil || m.ctx.Err() != nil { + return + } + + changes, err := watchNetworkChanges(m.ctx, m.l) + if err != nil { + // Not fatal. Everything else still works, we just won't notice a network change on our own. + m.l.Error("Failed to watch for network changes, will not rebind the udp listener when the network moves", + "error", err, + ) + return + } + + if changes == nil { + // This platform's sockets don't go stale, so there is nothing to watch for. + return + } + + m.l.Info("Watching for network changes to rebind the udp listener") + + for range changes { + m.l.Info("Local network changed, rebinding the udp listener") + rebind() + } +} diff --git a/udp/netchange_darwin.go b/udp/netchange_darwin.go new file mode 100644 index 00000000..483165e9 --- /dev/null +++ b/udp/netchange_darwin.go @@ -0,0 +1,164 @@ +//go:build darwin && !ios && !e2e_testing +// +build darwin,!ios,!e2e_testing + +package udp + +import ( + "context" + "encoding/binary" + "errors" + "log/slog" + "os" + "time" + + "golang.org/x/sys/unix" +) + +const ( + // netChangeSettleWindow is how long we keep swallowing routing messages after the first interesting one. A + // single network change is never a single message, it is a burst: the link drops, addresses go away, new ones + // arrive, routes get rewritten. Reporting part way through that just means reporting again. + netChangeSettleWindow = time.Second + + // netChangeReadBuffer is sized well past any rt_msghdr plus its addresses. A short read would be discarded by + // the kernel, so being generous here is how we avoid missing a message. + netChangeReadBuffer = 4096 +) + +// watchNetworkChanges reports when the local network moves out from under us, so the listener can be rebound. +// +// Darwin scopes a udp socket to whatever interface it came up on. Move between networks and we keep sending out an +// interface that no longer has a route, which surfaces as an instant "no route to host" with no packet ever leaving +// the box. Rebind clears that, but only if something notices the change and calls it. iOS has always been told by +// the host app off NWPathMonitor. This is the equivalent for everything else that runs on darwin. +// +// The returned channel is buffered and coalescing: a send is dropped if one is already pending, since both mean the +// same thing to a reader. It is closed when ctx is cancelled or the routing socket fails, so a caller can simply +// range over it. Platforms whose sockets do not need rebinding return a nil channel and no error. +func watchNetworkChanges(ctx context.Context, l *slog.Logger) (<-chan struct{}, error) { + sock, err := openRouteSocket() + if err != nil { + return nil, err + } + + changes := make(chan struct{}, 1) + + go func() { + defer close(changes) + defer func() { _ = sock.Close() }() + + // Closing the socket is what unblocks the read in watchRouteSocket, so this turns cancellation into a + // close. It is scoped to this call so it cannot outlive the watch it belongs to. + done := make(chan struct{}) + defer close(done) + go func() { + select { + case <-ctx.Done(): + _ = sock.Close() + case <-done: + } + }() + + watchRouteSocket(l, sock, changes) + }() + + return changes, nil +} + +// watchRouteSocket blocks reading the routing socket, reporting once per settled burst of changes. It returns when +// the socket is closed, which is how cancellation gets us out of here. +func watchRouteSocket(l *slog.Logger, sock *os.File, changes chan<- struct{}) { + buf := make([]byte, netChangeReadBuffer) + + for { + n, err := sock.Read(buf) + if err != nil { + logRouteSocketError(l, err) + return + } + + if !isNetworkChange(buf[:n]) { + continue + } + + // Swallow the rest of the burst. The deadline is absolute and not extended by what arrives, so this always + // ends after the settle window no matter how chatty the socket is. Changes that land after the window + // simply produce another report, which is the correct outcome anyway. + deadline := time.Now().Add(netChangeSettleWindow) + for { + if err = sock.SetReadDeadline(deadline); err != nil { + logRouteSocketError(l, err) + return + } + + if _, err = sock.Read(buf); err != nil { + if os.IsTimeout(err) { + break + } + logRouteSocketError(l, err) + return + } + } + + if err = sock.SetReadDeadline(time.Time{}); err != nil { + logRouteSocketError(l, err) + return + } + + select { + case changes <- struct{}{}: + default: + // One already pending, and a second "the network moved" tells the reader nothing new. + } + } +} + +// logRouteSocketError reports a routing socket failure unless it is just us shutting the socket down. +func logRouteSocketError(l *slog.Logger, err error) { + if errors.Is(err, os.ErrClosed) { + return + } + + l.Error("Error reading the routing socket, will no longer notice local network changes", "error", err) +} + +// openRouteSocket returns the routing socket as a non blocking os.File. Going through os.File puts reads on the go +// poller, which buys us both a working read deadline and a Close that unblocks a read in progress. +func openRouteSocket() (*os.File, error) { + fd, err := unix.Socket(unix.AF_ROUTE, unix.SOCK_RAW, unix.AF_UNSPEC) + if err != nil { + return nil, err + } + + if err = unix.SetNonblock(fd, true); err != nil { + _ = unix.Close(fd) + return nil, err + } + + return os.NewFile(uintptr(fd), "route"), nil +} + +// isNetworkChange reports whether a routing message means our local addressing may have moved out from under us. +// +// We read the header instead of parsing the message because the type is the only part we need, and a full parse can +// fail on shapes we don't care about, which would turn "a message I can't parse" into "a change I missed". +// rt_msghdr, if_msghdr and ifa_msghdr all begin with the same three fields, so this is the same for every type. +func isNetworkChange(msg []byte) bool { + if len(msg) < 4 { + return false + } + + // u_short msglen, u_char version, u_char type + if int(binary.NativeEndian.Uint16(msg[0:2])) > len(msg) || msg[2] != unix.RTM_VERSION { + return false + } + + switch msg[3] { + case unix.RTM_NEWADDR, unix.RTM_DELADDR, unix.RTM_IFINFO: + // An address arrived or left, or a link changed state. Anything else on this socket is either a route + // churning underneath us, which a rebind doesn't help with, or unrelated traffic. + return true + default: + return false + } +} diff --git a/udp/netchange_darwin_test.go b/udp/netchange_darwin_test.go new file mode 100644 index 00000000..a6d25c83 --- /dev/null +++ b/udp/netchange_darwin_test.go @@ -0,0 +1,244 @@ +//go:build darwin && !ios && !e2e_testing +// +build darwin,!ios,!e2e_testing + +package udp + +import ( + "context" + "encoding/binary" + "os" + "testing" + "time" + + "github.com/slackhq/nebula/config" + "github.com/slackhq/nebula/test" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.uber.org/goleak" + "golang.org/x/sys/unix" +) + +// routeMsg builds the first four bytes of a routing message, which is all isNetworkChange reads. +func routeMsg(msgType uint8, extra int) []byte { + msg := make([]byte, 4+extra) + binary.NativeEndian.PutUint16(msg[0:2], uint16(len(msg))) + msg[2] = unix.RTM_VERSION + msg[3] = msgType + return msg +} + +func TestIsNetworkChange(t *testing.T) { + // The three that mean our addressing may have moved + assert.True(t, isNetworkChange(routeMsg(unix.RTM_NEWADDR, 0))) + assert.True(t, isNetworkChange(routeMsg(unix.RTM_DELADDR, 0))) + assert.True(t, isNetworkChange(routeMsg(unix.RTM_IFINFO, 0))) + + // Route churn is not something a rebind helps with + assert.False(t, isNetworkChange(routeMsg(unix.RTM_ADD, 0))) + assert.False(t, isNetworkChange(routeMsg(unix.RTM_DELETE, 0))) + assert.False(t, isNetworkChange(routeMsg(unix.RTM_GET, 0))) + + // Garbage must not be mistaken for a change + assert.False(t, isNetworkChange(nil), "empty") + assert.False(t, isNetworkChange([]byte{0, 0, 0}), "short header") + + wrongVersion := routeMsg(unix.RTM_NEWADDR, 0) + wrongVersion[2] = unix.RTM_VERSION + 1 + assert.False(t, isNetworkChange(wrongVersion), "wrong rtm_version") + + lying := routeMsg(unix.RTM_NEWADDR, 0) + binary.NativeEndian.PutUint16(lying[0:2], 512) + assert.False(t, isNetworkChange(lying), "msglen longer than what we read") +} + +// socketPair returns a connected pair of datagram sockets, the first wrapped the same way the routing socket is. It +// stands in for the kernel so the watch loop can be driven with synthetic messages. +func socketPair(t *testing.T) (*os.File, int) { + t.Helper() + + fds, err := unix.Socketpair(unix.AF_UNIX, unix.SOCK_DGRAM, 0) + require.NoError(t, err) + require.NoError(t, unix.SetNonblock(fds[0], true)) + + f := os.NewFile(uintptr(fds[0]), "route") + t.Cleanup(func() { + _ = f.Close() + _ = unix.Close(fds[1]) + }) + + return f, fds[1] +} + +func TestWatchRouteSocketCoalescesABurst(t *testing.T) { + sock, kernel := socketPair(t) + changes := make(chan struct{}, 1) + + done := make(chan struct{}) + go func() { + watchRouteSocket(test.NewLogger(), sock, changes) + close(done) + }() + + // One network change is a burst of messages. All of these land inside the settle window, so they must produce + // exactly one report rather than one apiece. + for range 5 { + _, err := unix.Write(kernel, routeMsg(unix.RTM_NEWADDR, 8)) + require.NoError(t, err) + } + // Uninteresting messages in the middle of a burst must not add a report of their own either. + _, err := unix.Write(kernel, routeMsg(unix.RTM_ADD, 8)) + require.NoError(t, err) + + select { + case <-changes: + case <-time.After(netChangeSettleWindow * 4): + t.Fatal("a burst should have reported a change") + } + + // Nothing more from that burst + select { + case <-changes: + t.Fatal("a burst should report exactly once") + case <-time.After(netChangeSettleWindow): + } + + // A change after the window has closed is a separate event and gets its own report. + _, err = unix.Write(kernel, routeMsg(unix.RTM_IFINFO, 8)) + require.NoError(t, err) + select { + case <-changes: + case <-time.After(netChangeSettleWindow * 4): + t.Fatal("a later change should report again") + } + + // Closing the socket is how the real thing shuts down + require.NoError(t, sock.Close()) + select { + case <-done: + case <-time.After(time.Second * 5): + t.Fatal("watchRouteSocket did not return after the socket was closed") + } +} + +func TestWatchRouteSocketIgnoresUninterestingMessages(t *testing.T) { + sock, kernel := socketPair(t) + changes := make(chan struct{}, 1) + + done := make(chan struct{}) + go func() { + watchRouteSocket(test.NewLogger(), sock, changes) + close(done) + }() + + for _, msgType := range []uint8{unix.RTM_ADD, unix.RTM_DELETE, unix.RTM_GET, unix.RTM_MISS} { + _, err := unix.Write(kernel, routeMsg(msgType, 8)) + require.NoError(t, err) + } + + select { + case <-changes: + t.Fatal("route churn alone must not report a change") + case <-time.After(netChangeSettleWindow * 2): + } + + require.NoError(t, sock.Close()) + select { + case <-done: + case <-time.After(time.Second * 5): + t.Fatal("watchRouteSocket did not return after the socket was closed") + } +} + +// TestWatchRouteSocketDropsRatherThanBlocks covers the coalescing send. A reader that is busy rebinding must not +// wedge the watcher, and a second pending "the network moved" tells it nothing new anyway. +func TestWatchRouteSocketDropsRatherThanBlocks(t *testing.T) { + sock, kernel := socketPair(t) + changes := make(chan struct{}, 1) + + done := make(chan struct{}) + go func() { + watchRouteSocket(test.NewLogger(), sock, changes) + close(done) + }() + + // Nobody is reading changes, so after the first report the buffer is full for the rest of this test + for range 3 { + _, err := unix.Write(kernel, routeMsg(unix.RTM_NEWADDR, 8)) + require.NoError(t, err) + time.Sleep(netChangeSettleWindow + time.Millisecond*250) + } + + // The watcher must still be alive and responsive to a close + require.NoError(t, sock.Close()) + select { + case <-done: + case <-time.After(time.Second * 5): + t.Fatal("watchRouteSocket wedged on a full channel") + } + + assert.Len(t, changes, 1, "the pending report should have coalesced, not queued") +} + +// TestWatchNetworkChangesStopsWithContext covers the detection path against a real routing socket, including that +// cancelling the context closes the channel so a ranging caller falls out of its loop. +func TestWatchNetworkChangesStopsWithContext(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + + changes, err := watchNetworkChanges(ctx, test.NewLogger()) + require.NoError(t, err) + require.NotNil(t, changes, "darwin should support watching") + + drained := make(chan struct{}) + go func() { + for range changes { + } + close(drained) + }() + + cancel() + select { + case <-drained: + case <-time.After(time.Second * 5): + t.Fatal("cancelling the context should close the changes channel") + } +} + +// TestNetworkChangeMonitorStopsWithContext drives the whole monitor against a real routing socket: Start must block +// watching, and cancelling the context (which is all Control does on shutdown, it never stops the monitor directly) +// must return it and clean up the watch goroutines. +func TestNetworkChangeMonitorStopsWithContext(t *testing.T) { + // IgnoreCurrent because other tests in this package leave readers running; we only care about what this test + // leaks itself. + defer goleak.VerifyNone(t, goleak.IgnoreCurrent()) + + ctx, cancel := context.WithCancel(context.Background()) + + l := test.NewLogger() + c := config.NewC(l) + require.NoError(t, c.LoadString("listen:\n rebind_on_network_change: true\n")) + m := NewNetworkChangeMonitor(ctx, l, c) + + done := make(chan struct{}) + go func() { + m.Start(func() {}) + close(done) + }() + + // Start should be sitting on the routing socket, not have fallen out. If it returned early it either failed to + // watch or no-op'd, both of which we want to catch. + select { + case <-done: + t.Fatal("Start returned instead of watching") + case <-time.After(time.Millisecond * 250): + } + + cancel() + select { + case <-done: + case <-time.After(time.Second * 5): + t.Fatal("Start did not return after the context was cancelled") + } + + // Starting again after the context is dead must not open anything. + m.Start(func() {}) +} diff --git a/udp/netchange_generic.go b/udp/netchange_generic.go new file mode 100644 index 00000000..de5cc3a6 --- /dev/null +++ b/udp/netchange_generic.go @@ -0,0 +1,22 @@ +//go:build !darwin || ios || e2e_testing +// +build !darwin ios e2e_testing + +package udp + +import ( + "context" + "log/slog" +) + +// watchNetworkChanges is a no-op outside of darwin. +// +// Darwin is the platform that scopes a udp socket to the interface it came up on, so it is the platform whose socket +// goes stale when the local network changes. Everywhere else Rebind has nothing to do, so there is nothing to watch +// for. iOS is excluded on purpose even though it is darwin: the host app already drives the rebind off NWPathMonitor, +// and two things racing to rebind the same socket is worse than one. +// +// A nil channel means "not supported here", which callers must treat as "do not start a watcher" rather than +// selecting on it, since a receive from a nil channel blocks forever. +func watchNetworkChanges(_ context.Context, _ *slog.Logger) (<-chan struct{}, error) { + return nil, nil +} diff --git a/udp/netchange_test.go b/udp/netchange_test.go new file mode 100644 index 00000000..fe57a2e2 --- /dev/null +++ b/udp/netchange_test.go @@ -0,0 +1,39 @@ +package udp + +import ( + "context" + "testing" + + "github.com/slackhq/nebula/config" + "github.com/slackhq/nebula/test" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func newMonitor(t *testing.T, ctx context.Context, cfg string) *NetworkChangeMonitor { + t.Helper() + l := test.NewLogger() + c := config.NewC(l) + require.NoError(t, c.LoadString(cfg)) + return NewNetworkChangeMonitor(ctx, l, c) +} + +func TestNetworkChangeMonitorDefaultsOn(t *testing.T) { + // Says nothing about rebinding, so this covers the default. + m := newMonitor(t, context.Background(), "listen:\n host: 0.0.0.0\n") + assert.True(t, m.enabled, "should default to on") +} + +func TestNetworkChangeMonitorDisabledIsANoOp(t *testing.T) { + m := newMonitor(t, context.Background(), "listen:\n rebind_on_network_change: false\n") + require.False(t, m.enabled) + + // Must return without opening a socket. If it watched anything this would block. + m.Start(func() {}) +} + +func TestNetworkChangeMonitorNilRebindIsANoOp(t *testing.T) { + // Nothing to rebind, so there is no point watching, on any platform. + m := newMonitor(t, context.Background(), "listen:\n rebind_on_network_change: true\n") + m.Start(nil) +} diff --git a/udp/udp_darwin.go b/udp/udp_darwin.go index 3d6b39a5..574e4494 100644 --- a/udp/udp_darwin.go +++ b/udp/udp_darwin.go @@ -187,6 +187,9 @@ func (u *StdConn) SupportsMultipleReaders() bool { return false } +// Rebind clears the interface the kernel scoped this socket to, so that sends are routed against the current +// routing table instead of the interface we happened to be on when the socket was created. Darwin pins sockets +// this way on its own, which is what strands us after the underlying network changes. func (u *StdConn) Rebind() error { var err error if u.isV4 { @@ -195,9 +198,5 @@ func (u *StdConn) Rebind() error { err = syscall.SetsockoptInt(int(u.sysFd), syscall.IPPROTO_IPV6, syscall.IPV6_BOUND_IF, 0) } - if err != nil { - u.l.Error("Failed to rebind udp socket", "error", err) - } - - return nil + return err } diff --git a/udp/udp_tester.go b/udp/udp_tester.go index f872e32a..9c0d989f 100644 --- a/udp/udp_tester.go +++ b/udp/udp_tester.go @@ -10,6 +10,7 @@ import ( "net/netip" "os" "sync" + "sync/atomic" "github.com/slackhq/nebula/config" "github.com/slackhq/nebula/header" @@ -64,7 +65,9 @@ func acquirePacket() *Packet { } type TesterConn struct { - Addr netip.AddrPort + // addr is read by nebula's own goroutines on every send and by the router's flow renderer, and a test can + // move it mid-run to simulate roaming, so it is atomic rather than a plain field. + addr atomic.Pointer[netip.AddrPort] RxPackets chan *Packet // Packets to receive into nebula TxPackets chan *Packet // Packets transmitted outside by nebula @@ -82,13 +85,24 @@ type TesterConn struct { } func NewListener(l *slog.Logger, ip netip.Addr, port int, _ bool, _ int) (Conn, error) { - return &TesterConn{ - Addr: netip.AddrPortFrom(ip, uint16(port)), + c := &TesterConn{ RxPackets: make(chan *Packet, 10), TxPackets: make(chan *Packet, 10), done: make(chan struct{}), l: l, - }, nil + } + c.SetAddr(netip.AddrPortFrom(ip, uint16(port))) + return c, nil +} + +// GetAddr returns the underlay address this conn currently sends from. +func (u *TesterConn) GetAddr() netip.AddrPort { + return *u.addr.Load() +} + +// SetAddr moves this conn to a new underlay address, standing in for a host waking up on a different network. +func (u *TesterConn) SetAddr(addr netip.AddrPort) { + u.addr.Store(&addr) } // Send will place a UdpPacket onto the receive queue for nebula to consume @@ -147,7 +161,7 @@ func (u *TesterConn) WriteTo(b []byte, addr netip.AddrPort) error { p.Data = p.Data[:len(b)] } copy(p.Data, b) - p.From = u.Addr + p.From = u.GetAddr() p.To = addr select { case <-u.done: @@ -178,7 +192,7 @@ func NewUDPStatsEmitter(_ []Conn) func() { } func (u *TesterConn) LocalAddr() (netip.AddrPort, error) { - return u.Addr, nil + return u.GetAddr(), nil } func (u *TesterConn) SupportsMultipleReaders() bool {