nebula/udp/netchange_darwin_test.go

//go: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() {})
}