diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index ca2e45e0..b22b3430 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -12,7 +12,7 @@ jobs: steps: - uses: actions/checkout@v7 - - uses: actions/setup-go@v6 + - uses: actions/setup-go@v7 with: go-version: '1.26' check-latest: true @@ -38,7 +38,7 @@ jobs: steps: - uses: actions/checkout@v7 - - uses: actions/setup-go@v6 + - uses: actions/setup-go@v7 with: go-version: '1.26' check-latest: true @@ -78,7 +78,7 @@ jobs: steps: - uses: actions/checkout@v7 - - uses: actions/setup-go@v6 + - uses: actions/setup-go@v7 with: go-version: '1.26' check-latest: true diff --git a/.github/workflows/smoke-extra.yml b/.github/workflows/smoke-extra.yml index b15dff4e..445b83ed 100644 --- a/.github/workflows/smoke-extra.yml +++ b/.github/workflows/smoke-extra.yml @@ -32,7 +32,7 @@ jobs: - uses: actions/checkout@v7 - - uses: actions/setup-go@v6 + - uses: actions/setup-go@v7 with: go-version: '1.26' check-latest: true @@ -64,7 +64,7 @@ jobs: - uses: actions/checkout@v7 - - uses: actions/setup-go@v6 + - uses: actions/setup-go@v7 with: go-version: '1.26' check-latest: true @@ -90,7 +90,7 @@ jobs: - uses: actions/checkout@v7 - - uses: actions/setup-go@v6 + - uses: actions/setup-go@v7 with: go-version: '1.26' check-latest: true diff --git a/.github/workflows/smoke.yml b/.github/workflows/smoke.yml index 82d06385..dc274932 100644 --- a/.github/workflows/smoke.yml +++ b/.github/workflows/smoke.yml @@ -20,7 +20,7 @@ jobs: - uses: actions/checkout@v7 - - uses: actions/setup-go@v6 + - uses: actions/setup-go@v7 with: go-version: '1.26' check-latest: true diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 447d4870..eab4e4c6 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -20,7 +20,7 @@ jobs: - uses: actions/checkout@v7 - - uses: actions/setup-go@v6 + - uses: actions/setup-go@v7 with: go-version: '1.26' check-latest: true @@ -85,7 +85,7 @@ jobs: - uses: actions/checkout@v7 - - uses: actions/setup-go@v6 + - uses: actions/setup-go@v7 with: go-version: '1.26' check-latest: true @@ -130,7 +130,7 @@ jobs: - uses: actions/checkout@v7 - - uses: actions/setup-go@v6 + - uses: actions/setup-go@v7 with: go-version: '1.26' check-latest: true 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/connection_state.go b/connection_state.go index 0ae2d9be..0d6f59e1 100644 --- a/connection_state.go +++ b/connection_state.go @@ -2,11 +2,13 @@ package nebula import ( "encoding/json" + "log/slog" "sync" "sync/atomic" "github.com/slackhq/nebula/cert" "github.com/slackhq/nebula/handshake" + "github.com/slackhq/nebula/header" "github.com/slackhq/nebula/noiseutil" ) @@ -20,6 +22,7 @@ type ConnectionState struct { initiator bool messageCounter atomic.Uint64 window *Bits + decryptLock sync.Mutex writeLock sync.Mutex } @@ -54,3 +57,52 @@ func (cs *ConnectionState) MarshalJSON() ([]byte, error) { func (cs *ConnectionState) Curve() cert.Curve { return cs.myCert.Curve() } + +func (cs *ConnectionState) Decrypt(l *slog.Logger, messageCounter uint64, out []byte, packet []byte, nb []byte) ([]byte, error) { + var err error + cs.decryptLock.Lock() + result := cs.window.Check(l, messageCounter) + cs.decryptLock.Unlock() + if !result { + return nil, ErrAlreadySeen + } + + out, err = cs.dKey.DecryptDanger(out, packet[:header.Len], packet[header.Len:], messageCounter, nb) + if err != nil { + return nil, err + } + + cs.decryptLock.Lock() + result = cs.window.Update(l, messageCounter) + cs.decryptLock.Unlock() + if !result { + return nil, ErrAlreadySeen + } + return out, nil +} + +// VerifyRelay verifies AEAD protected (but not encrypted) relay frames. packet must be length-checked by the caller. +func (cs *ConnectionState) VerifyRelay(l *slog.Logger, messageCounter uint64, packet []byte, nb []byte) error { + cs.decryptLock.Lock() + result := cs.window.Check(l, messageCounter) + cs.decryptLock.Unlock() + if !result { + return ErrAlreadySeen + } + + signedPayload := packet[:len(packet)-cs.dKey.Overhead()] + signatureValue := packet[len(packet)-cs.dKey.Overhead():] + _, err := cs.dKey.DecryptDanger(nil, signedPayload, signatureValue, messageCounter, nb) + if err != nil { + return err + } + + cs.decryptLock.Lock() + result = cs.window.Update(l, messageCounter) + cs.decryptLock.Unlock() + if !result { + return ErrAlreadySeen + } + + return nil +} 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/dns_server.go b/dns_server.go index a80630b5..9339b068 100644 --- a/dns_server.go +++ b/dns_server.go @@ -97,8 +97,7 @@ func (d *dnsServer) reload(c *config.C, initial bool) error { newAddr := getDnsServerAddr(c) d.serverMu.Lock() - running := d.server - runningStarted := d.started + running := d.server != nil sameAddr := d.addr == newAddr d.addr = newAddr d.enabled.Store(enabled) @@ -112,7 +111,7 @@ func (d *dnsServer) reload(c *config.C, initial bool) error { } if !enabled { - if running != nil { + if running { d.Stop() } // Drop any records that accumulated while enabled; a later re-enable @@ -121,12 +120,12 @@ func (d *dnsServer) reload(c *config.C, initial bool) error { return nil } - if running == nil { + if !running { // Was disabled (or never started); bring it up now. go d.Start() } else if !sameAddr { - d.shutdownServer(running, runningStarted, "reload") - // Old Start goroutine has now exited; bring up a fresh listener on the new address. + // Stop clears the slot before shutting down, otherwise the Start below can find the dying server and refuse + d.Stop() go d.Start() } @@ -162,7 +161,9 @@ func (d *dnsServer) Start() { started := make(chan struct{}) d.serverMu.Lock() - if d.ctx.Err() != nil { + // Re-check enabled under the lock, a disable that raced our check above snapshots the slot under it too. + // Two reloads in quick succession can both spawn a Start, the loser would orphan the live listener past Stop + if d.ctx.Err() != nil || d.server != nil || !d.enabled.Load() { d.serverMu.Unlock() return } @@ -200,6 +201,14 @@ func (d *dnsServer) Start() { close(started) } + // Release our slot, unless a reload already replaced us, so a dead listener can't block a future Start + d.serverMu.Lock() + if d.server == server { + d.server = nil + d.started = nil + } + d.serverMu.Unlock() + if err != nil { d.l.Warn("Failed to run the DNS responder", "error", err) } diff --git a/dns_server_test.go b/dns_server_test.go index 58646937..73267db2 100644 --- a/dns_server_test.go +++ b/dns_server_test.go @@ -194,14 +194,51 @@ func TestDnsServer_reload_initial_serveDnsWithoutLighthouse(t *testing.T) { } func TestDnsServer_reload_sameAddr_noOp(t *testing.T) { + port := freeUDPPort(t) ds, c := newTestDnsServer(t) - setDnsConfig(c, "127.0.0.1", "0", true, true) - + setDnsConfig(c, "127.0.0.1", port, true, true) require.NoError(t, ds.reload(c, true)) - // No server running yet, no addr change. Reload should not spawn anything. + + go ds.Start() + waitForBind(t, ds) + + ds.serverMu.Lock() + before := ds.server + ds.serverMu.Unlock() + require.NotNil(t, before) + + // Same address, so the running listener must be left alone rather than rebuilt under live queries require.NoError(t, ds.reload(c, false)) assert.True(t, ds.enabled.Load()) - assert.Nil(t, ds.server) + + ds.serverMu.Lock() + after := ds.server + ds.serverMu.Unlock() + assert.Same(t, before, after, "a same-address reload must not restart the listener") + + ds.Stop() +} + +// The branch the old sameAddr test was accidentally hitting: enabled with nothing running means reload starts it. +func TestDnsServer_reload_whenNotRunning_starts(t *testing.T) { + port := freeUDPPort(t) + ds, c := newTestDnsServer(t) + setDnsConfig(c, "127.0.0.1", port, true, true) + + // initial only records config, it never starts anything + require.NoError(t, ds.reload(c, true)) + ds.serverMu.Lock() + assert.Nil(t, ds.server, "the initial reload must not start a listener") + ds.serverMu.Unlock() + + require.NoError(t, ds.reload(c, false)) + waitForBind(t, ds) + + ds.serverMu.Lock() + assert.NotNil(t, ds.server, "a reload with nothing running should bring DNS up") + ds.serverMu.Unlock() + + ds.Stop() } func TestDnsServer_StartStop_lifecycle(t *testing.T) { @@ -427,3 +464,168 @@ func waitFor(t *testing.T, cond func() bool) { } t.Fatal("timed out waiting for condition") } + +// Two reloads in quick succession, or a HUP before Control.Start, can race two Starts at the same listener. +func TestDnsServer_Start_isIdempotent(t *testing.T) { + port := freeUDPPort(t) + ds, c := newTestDnsServer(t) + setDnsConfig(c, "127.0.0.1", port, true, true) + require.NoError(t, ds.reload(c, true)) + + go ds.Start() + waitForBind(t, ds) + + ds.serverMu.Lock() + first := ds.server + ds.serverMu.Unlock() + require.NotNil(t, first) + + // If the second Start replaces the tracked server, Stop kills the wrong one and the port leaks + done := make(chan struct{}) + go func() { + ds.Start() + close(done) + }() + select { + case <-done: + case <-time.After(time.Second * 5): + t.Fatal("second Start never returned") + } + + ds.serverMu.Lock() + second := ds.server + ds.serverMu.Unlock() + assert.Same(t, first, second, "a second Start must not replace the running server") + + // The real proof, after Stop the port must actually be free + ds.Stop() + waitFor(t, func() bool { + pc, err := net.ListenPacket("udp", "127.0.0.1:"+port) + if err != nil { + return false + } + _ = pc.Close() + return true + }) +} + +// An address change must actually end up listening on the new port. Start's guard refuses when a server is already +// installed, so reload has to clear the slot before shutting the old one down. +func TestDnsServer_reload_addrChange_restarts(t *testing.T) { + first := freeUDPPort(t) + second := freeUDPPort(t) + + ds, c := newTestDnsServer(t) + setDnsConfig(c, "127.0.0.1", first, true, true) + require.NoError(t, ds.reload(c, true)) + + go ds.Start() + waitForBind(t, ds) + + // Cycle a few times, the failure this guards against depends on which goroutine wins serverMu + for i := range 8 { + want := second + if i%2 == 1 { + want = first + } + setDnsConfig(c, "127.0.0.1", want, true, true) + require.NoError(t, ds.reload(c, false)) + waitForBind(t, ds) + + ds.serverMu.Lock() + srv := ds.server + ds.serverMu.Unlock() + require.NotNil(t, srv, "reload left DNS down instead of restarting it") + require.Equal(t, "127.0.0.1:"+want, srv.Addr, "reload should be serving the new address") + } + + // Land back on second so the port assertions below are meaningful + setDnsConfig(c, "127.0.0.1", second, true, true) + require.NoError(t, ds.reload(c, false)) + waitForBind(t, ds) + + // The old port must be released and the new one actually held + waitFor(t, func() bool { + pc, err := net.ListenPacket("udp", "127.0.0.1:"+first) + if err != nil { + return false + } + _ = pc.Close() + return true + }) + _, err := net.ListenPacket("udp", "127.0.0.1:"+second) + require.Error(t, err, "the new address should be bound by the DNS responder") + + ds.Stop() +} + +// A listener that dies on its own must release the slot, or a later same-addr reload sees it as running and no-ops. +func TestDnsServer_Start_bindFailure_releasesSlot(t *testing.T) { + port := freeUDPPort(t) + blocker, err := net.ListenPacket("udp", "127.0.0.1:"+port) + require.NoError(t, err) + + ds, c := newTestDnsServer(t) + setDnsConfig(c, "127.0.0.1", port, true, true) + require.NoError(t, ds.reload(c, true)) + + ds.Start() // returns once the bind fails + + ds.serverMu.Lock() + assert.Nil(t, ds.server, "a listener that failed to bind must not stay parked in the slot") + ds.serverMu.Unlock() + + // With the slot released, a reload can retry once the port frees up + require.NoError(t, blocker.Close()) + require.NoError(t, ds.reload(c, false)) + waitForBind(t, ds) + + ds.serverMu.Lock() + assert.NotNil(t, ds.server, "a same-addr reload should retry after a failed bind") + ds.serverMu.Unlock() + + ds.Stop() +} + +// A disable that lands while Start is between its unlocked check and the guard must not leave a listener behind. +func TestDnsServer_Start_refusesWhenDisabledUnderLock(t *testing.T) { + port := freeUDPPort(t) + ds, c := newTestDnsServer(t) + setDnsConfig(c, "127.0.0.1", port, true, true) + require.NoError(t, ds.reload(c, true)) + require.True(t, ds.enabled.Load()) + + // Holding serverMu parks Start on the lock, the only way to land the disable in that window on purpose + ds.serverMu.Lock() + + done := make(chan struct{}) + go func() { + ds.Start() + close(done) + }() + + select { + case <-done: + ds.serverMu.Unlock() + t.Fatal("Start returned early, the test never exercised the window") + case <-time.After(time.Millisecond * 100): + } + + // The disable reload's critical section. It sees nothing running, so it never calls Stop. + ds.enabled.Store(false) + ds.serverMu.Unlock() + + select { + case <-done: + case <-time.After(time.Second * 5): + t.Fatal("Start never returned") + } + + ds.serverMu.Lock() + assert.Nil(t, ds.server, "Start must not install a listener a disable already cancelled") + ds.serverMu.Unlock() + + pc, err := net.ListenPacket("udp", "127.0.0.1:"+port) + require.NoError(t, err, "an orphaned listener is still holding the port") + _ = pc.Close() +} 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/examples/service_scripts/nebula.service b/examples/service_scripts/nebula.service index ab5218f8..295a6fcd 100644 --- a/examples/service_scripts/nebula.service +++ b/examples/service_scripts/nebula.service @@ -8,6 +8,15 @@ Before=sshd.service Type=notify NotifyAccess=main SyslogIdentifier=nebula + +# Uncomment to run as an unprivileged user with only CAP_NET_ADMIN. Requires a +# nebula user that owns the config directory. Add CAP_NET_BIND_SERVICE to both +# lines if any listener (lighthouse DNS, listen.port, stats, sshd) binds <1024. +#User=nebula +#Group=nebula +#CapabilityBoundingSet=CAP_NET_ADMIN +#AmbientCapabilities=CAP_NET_ADMIN + ExecReload=/bin/kill -HUP $MAINPID ExecStart=/usr/local/bin/nebula -config /etc/nebula/config.yml Restart=always diff --git a/go.mod b/go.mod index 32aa5650..9748492d 100644 --- a/go.mod +++ b/go.mod @@ -1,6 +1,6 @@ module github.com/slackhq/nebula -go 1.26 +go 1.26.0 require ( dario.cat/mergo v1.0.2 @@ -24,12 +24,12 @@ require ( github.com/vishvananda/netlink v1.3.1 go.uber.org/goleak v1.3.0 go.yaml.in/yaml/v3 v3.0.4 - golang.org/x/crypto v0.53.0 + golang.org/x/crypto v0.54.0 golang.org/x/exp v0.0.0-20230725093048-515e97ebf090 - golang.org/x/net v0.56.0 - golang.org/x/sync v0.21.0 - golang.org/x/sys v0.46.0 - golang.org/x/term v0.44.0 + golang.org/x/net v0.57.0 + golang.org/x/sync v0.22.0 + golang.org/x/sys v0.47.0 + golang.org/x/term v0.45.0 golang.zx2c4.com/wintun v0.0.0-20230126152724-0fa3db229ce2 golang.zx2c4.com/wireguard v0.0.0-20230325221338-052af4a8072b golang.zx2c4.com/wireguard/windows v1.0.1 diff --git a/go.sum b/go.sum index 11e72276..29d68429 100644 --- a/go.sum +++ b/go.sum @@ -162,8 +162,8 @@ golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACk golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= golang.org/x/crypto v0.0.0-20210322153248-0c34fe9e7dc2/go.mod h1:T9bdIzuCu7OtxOm1hfPfRQxPLYneinmdGuTeoZ9dtd4= -golang.org/x/crypto v0.53.0 h1:QZ4Muo8THX6CizN2vPPd5fBGHyogrdK9fG4wLPFUsto= -golang.org/x/crypto v0.53.0/go.mod h1:DNLU434OwVakk9PzuwV8w62mAJpRJL3vsgcfp4Qnsio= +golang.org/x/crypto v0.54.0 h1:YLIA59K4fiNzHzjnZt2tUJQjQtUWfWbeHBqKtk3eScw= +golang.org/x/crypto v0.54.0/go.mod h1:KWL8ny2AZdGR2cWmzeHrp2azQPGogOv+HeQaVEXC2dk= golang.org/x/exp v0.0.0-20230725093048-515e97ebf090 h1:Di6/M8l0O2lCLc6VVRWhgCiApHV8MnQurBnFSHsQtNY= golang.org/x/exp v0.0.0-20230725093048-515e97ebf090/go.mod h1:FXUEEKJgO7OQYeo8N01OfiKP8RXMtf6e8aTskBGqWdc= golang.org/x/lint v0.0.0-20200302205851-738671d3881b/go.mod h1:3xt1FjdF8hUf6vQPIChWIBhFzV8gjjsPE/fR3IyQdNY= @@ -182,8 +182,8 @@ golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLL golang.org/x/net v0.0.0-20200625001655-4c5254603344/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA= golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU= golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg= -golang.org/x/net v0.56.0 h1:Rw8j/hFzGvJUZwNBXnAtf5sVDVt+65SK2C7IxCxZt5o= -golang.org/x/net v0.56.0/go.mod h1:D3Ku6r+V6JROoZK144D2XfMHFcMq/0zSfLelVTCFKec= +golang.org/x/net v0.57.0 h1:K5+3DljvIuDG9/Jv9rvyMywYNFCQ9RSUY6OOTTkT+tE= +golang.org/x/net v0.57.0/go.mod h1:KpXc8iv+r3XplLAG/f7Jsf9RPszJzdR0f58q9vGOuEU= golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw= golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= @@ -191,8 +191,8 @@ golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJ golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20201207232520-09787c993a3a/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= -golang.org/x/sync v0.21.0 h1:HLII4xRRTtCRkxYp4HNFF0Js/Og6q2i++KXbg0gHCwM= -golang.org/x/sync v0.21.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= +golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek= +golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20181116152217-5ac8a444bdc5/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= @@ -208,11 +208,11 @@ golang.org/x/sys v0.0.0-20210124154548-22da62e12c0c/go.mod h1:h1NjWce9XRLGQEsW7w golang.org/x/sys v0.0.0-20210603081109-ebe580a85c40/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.2.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.10.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.46.0 h1:noSf2Fq6F8DBgS+LysIkx7rIExoNHJsxOAtPp4rthXw= -golang.org/x/sys v0.46.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs= +golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= -golang.org/x/term v0.44.0 h1:0rLvDRCtNj0gZkyIXhCyOb2OAzEhLVqc4B+hrsBhrmc= -golang.org/x/term v0.44.0/go.mod h1:7ze4MdzUzLXpSAoFP1H0bOI9aXDqveSvatT5vKcFh2Y= +golang.org/x/term v0.45.0 h1:NwWyBmoJCbfTHpxrWoZ9C6/VxOf7ic219I8xZZFdrf0= +golang.org/x/term v0.45.0/go.mod h1:9aqxs0blBcrm/n0L9QW0aRVD+ktan8ssZromtqJC43w= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk= golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= diff --git a/handshake_manager.go b/handshake_manager.go index 6a2d0b4a..b4bebdc7 100644 --- a/handshake_manager.go +++ b/handshake_manager.go @@ -295,7 +295,13 @@ func (hm *HandshakeManager) handleOutbound(vpnIp netip.Addr, lighthouseTriggered hm.messageMetrics.Tx(header.Handshake, hh.machine.Subtype(), 1) err := hm.outside.WriteTo(stage0, addr) if err != nil { - hostinfo.logger(hm.l).Error("Failed to send handshake message", + // These repeat every attempt, so match the success log below and only shout when the remotes changed + level := slog.LevelDebug + if remotesHaveChanged { + level = slog.LevelError + } + + hostinfo.logger(hm.l).Log(context.Background(), level, "Failed to send handshake message", "udpAddr", addr, "initiatorIndex", hostinfo.localIndexId, "handshake", hsFields, @@ -529,7 +535,9 @@ func (hm *HandshakeManager) DeleteHostInfo(hostinfo *HostInfo) { func (hm *HandshakeManager) unlockedDeleteHostInfo(hostinfo *HostInfo) { for _, addr := range hostinfo.vpnAddrs { - delete(hm.vpnIps, addr) + if cur, ok := hm.vpnIps[addr]; ok && cur.hostinfo == hostinfo { + delete(hm.vpnIps, addr) + } } if len(hm.vpnIps) == 0 { diff --git a/inside.go b/inside.go index 163a6034..a80b2e96 100644 --- a/inside.go +++ b/inside.go @@ -408,7 +408,7 @@ func (f *Interface) sendNoMetrics(t header.MessageType, st header.MessageSubType if err != nil { hostinfo.logger(f.l).Error("Failed to write outgoing packet", "error", err, - "udpAddr", remote, + "udpAddr", hr, ) } } else { 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/outside.go b/outside.go index 8e89f807..cf56bb4a 100644 --- a/outside.go +++ b/outside.go @@ -102,27 +102,31 @@ func (f *Interface) readOutsidePackets(via ViaSender, out []byte, packet []byte, return } + if len(packet) < header.Len+hostinfo.ConnectionState.dKey.Overhead() { + f.messageMetrics.RxInvalid(1) + if f.l.Enabled(context.Background(), slog.LevelDebug) { + f.l.Debug("packet too small", "from", via, "length", len(packet)) + } + return + } + // All remaining packets are encrypted - ci := hostinfo.ConnectionState - if !ci.window.Check(f.l, h.MessageCounter) { - return - } - - // Relay packets are special if isMessageRelay { + // Relay packets are special, this branch should always early-return + if err = hostinfo.ConnectionState.VerifyRelay(f.l, h.MessageCounter, packet, nb); err != nil { + if f.l.Enabled(context.Background(), slog.LevelDebug) { + hostinfo.logger(f.l).Debug("Failed to verify relay packet", "error", err, "from", via, "header", h) + } + return + } f.handleOutsideRelayPacket(hostinfo, via, out, packet, h, fwPacket, lhf, nb, q, localCache) - return } - out, err = f.decrypt(hostinfo, h.MessageCounter, out, packet, h, nb) + out, err = hostinfo.ConnectionState.Decrypt(f.l, h.MessageCounter, out, packet, 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, - ) + hostinfo.logger(f.l).Debug("Failed to decrypt packet", "error", err, "from", via, "header", h) } return } @@ -151,7 +155,7 @@ func (f *Interface) readOutsidePackets(via ViaSender, out []byte, packet []byte, // No-op, useful for the Roaming and connectionManager side-effects above case header.TestRequest: //recycle the input packet ciphertext as our output buffer - f.send(header.Test, header.TestReply, ci, hostinfo, out, nb, packet) + f.send(header.Test, header.TestReply, hostinfo.ConnectionState, hostinfo, out, nb, packet) default: hostinfo.logger(f.l).Error("IsValidSubType was true, but unexpected test subtype seen", "from", via, "header", h) return @@ -170,27 +174,8 @@ func (f *Interface) readOutsidePackets(via ViaSender, out []byte, packet []byte, } func (f *Interface) handleOutsideRelayPacket(hostinfo *HostInfo, via ViaSender, out []byte, packet []byte, h *header.H, fwPacket *firewall.Packet, lhf *LightHouseHandler, nb []byte, q int, localCache firewall.ConntrackCache) { - // The entire body is sent as AD, not encrypted. - // The packet consists of a 16-byte parsed Nebula header, Associated Data-protected payload, and a trailing 16-byte AEAD signature value. - // The packet is guaranteed to be at least 16 bytes at this point, b/c it got past the h.Parse() call above. If it's - // otherwise malformed (meaning, there is no trailing 16 byte AEAD value), then this will result in at worst a 0-length slice - // which will gracefully fail in the DecryptDanger call. - signedPayload := packet[:len(packet)-hostinfo.ConnectionState.dKey.Overhead()] - signatureValue := packet[len(packet)-hostinfo.ConnectionState.dKey.Overhead():] - var err error - out, err = hostinfo.ConnectionState.dKey.DecryptDanger(out, signedPayload, signatureValue, h.MessageCounter, nb) - if err != nil { - return - } - // Advance the replay window now that the frame is authenticated - if !hostinfo.ConnectionState.window.Update(f.l, h.MessageCounter) { - if f.l.Enabled(context.Background(), slog.LevelDebug) { - hostinfo.logger(f.l).Debug("dropping out of window relay packet", "header", h) - } - return - } - // Successfully validated the thing. Get rid of the Relay header. - signedPayload = signedPayload[header.Len:] + // Successfully validated the thing. Get rid of the Relay header and the AEAD tag + signedPayload := packet[header.Len : len(packet)-hostinfo.ConnectionState.dKey.Overhead()] // Pull the Roaming parts up here, and return in all call paths. f.handleHostRoaming(hostinfo, via) // Track usage of both the HostInfo and the Relay for the received & authenticated packet @@ -234,9 +219,10 @@ func (f *Interface) handleOutsideRelayPacket(hostinfo *HostInfo, via ViaSender, if targetRelay.State == Established { switch targetRelay.Type { case ForwardingType: - // Forward this packet through the relay tunnel - // Find the target HostInfo - f.SendVia(targetHI, targetRelay, signedPayload, nb, out, false) + // Forward this packet through the relay tunnel, rebuilding it in place. + // Encode overwrites the old outer header, and the new AEAD tag lands where the old one was + fwdBuf := packet[:0:len(packet)] // Cap to len(packet) to protect memory from a larger parent buffer + f.SendVia(targetHI, targetRelay, signedPayload, nb, fwdBuf, true) case TerminalType: hostinfo.logger(f.l).Error("Unexpected Relay Type of Terminal") return @@ -503,20 +489,6 @@ func parseV4(data []byte, incoming bool, fp *firewall.Packet) error { return nil } -func (f *Interface) decrypt(hostinfo *HostInfo, mc uint64, out []byte, packet []byte, h *header.H, nb []byte) ([]byte, error) { - var err error - out, err = hostinfo.ConnectionState.dKey.DecryptDanger(out, packet[:header.Len], packet[header.Len:], mc, nb) - if err != nil { - return nil, err - } - - if !hostinfo.ConnectionState.window.Update(f.l, mc) { - return nil, ErrOutOfWindow - } - - return out, nil -} - func (f *Interface) handleOutsideMessagePacket(hostinfo *HostInfo, out []byte, packet []byte, fwPacket *firewall.Packet, nb []byte, q int, localCache firewall.ConntrackCache) { err := newPacket(out, true, fwPacket) if err != 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 {