fix: prevent tailnet coordination panics (#29300) (#29743)

Backport of #29300 to `release/2.34` for
[PLAT-595](https://linear.app/codercom/issue/PLAT-595).

Includes the full fix and regression tests for malformed coordinate
requests, coordinator panic containment, and disconnect cleanup. The
conflict in `tailnet/tunnel.go` preserves the release branch's existing
tunnel check and adds the upstream validation unchanged.

enterprise/tailnet/connio.go

@@ -3,6 +3,7 @@ package tailnet
 import (
 	"context"
 	"fmt"
+	"runtime/debug"
 	"slices"
 	"sync"
 
@@ -100,6 +101,17 @@ func (c *connIO) recvLoop() {
 		}
 	}()
 	defer c.Close()
+	// This must be deferred after Close so it runs first: Enqueue needs the
+	// response channel open to deliver CloseErrInternal to the peer.
+	defer func() {
+		if recovered := recover(); recovered != nil {
+			c.logger.Error(c.peerCtx, "panic handling peer request (recovered)",
+				slog.F("panic", recovered),
+				slog.F("stack", string(debug.Stack())),
+			)
+			_ = c.Enqueue(&proto.CoordinateResponse{Error: agpl.CloseErrInternal})
+		}
+	}()
 	for {
 		select {
 		case <-c.coordCtx.Done():
@@ -128,6 +140,10 @@ var errDisconnect = xerrors.New("graceful disconnect")
 
 func (c *connIO) handleRequest(req *proto.CoordinateRequest) error {
 	c.logger.Debug(c.peerCtx, "got request")
+	if err := agpl.ValidateCoordinateRequest(req); err != nil {
+		c.logger.Warn(c.peerCtx, "invalid coordinate request", slog.Error(err))
+		return err
+	}
 	err := c.auth.Authorize(c.peerCtx, req)
 	if err != nil {
 		c.logger.Warn(c.peerCtx, "unauthorized request", slog.Error(err))

enterprise/tailnet/connio_internal_test.go

@@ -0,0 +1,89 @@
+package tailnet
+
+import (
+	"context"
+	"strings"
+	"testing"
+
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/require"
+
+	"cdr.dev/slog/v3/sloggers/sloghuman"
+	"cdr.dev/slog/v3/sloggers/slogtest"
+	agpl "github.com/coder/coder/v2/tailnet"
+	"github.com/coder/coder/v2/tailnet/proto"
+	agpltest "github.com/coder/coder/v2/tailnet/test"
+	"github.com/coder/coder/v2/testutil"
+)
+
+func TestConnIOHandleRequestRejectsBeforeMutation(t *testing.T) {
+	t.Parallel()
+	ctx := testutil.Context(t, testutil.WaitShort)
+	bindings := make(chan binding, 1)
+	var logbuf strings.Builder
+	c := &connIO{
+		id:       uuid.New(),
+		coordCtx: ctx,
+		peerCtx:  ctx,
+		logger:   slogtest.Make(t, &slogtest.Options{IgnoreErrors: true}).AppendSinks(sloghuman.Sink(&logbuf)),
+		bindings: bindings,
+		auth:     agpl.SingleTailnetCoordinateeAuth{},
+	}
+
+	err := c.handleRequest(&proto.CoordinateRequest{
+		UpdateSelf: &proto.CoordinateRequest_UpdateSelf{Node: &proto.Node{
+			PreferredDerp: 31002,
+		}},
+		ReadyForHandshake: []*proto.CoordinateRequest_ReadyForHandshake{nil},
+	})
+	require.EqualError(t, err, "ready_for_handshake entries must not be nil")
+	require.Contains(t, logbuf.String(), "invalid coordinate request")
+	select {
+	case binding := <-bindings:
+		t.Fatalf("unexpected binding: %+v", binding)
+	default:
+	}
+}
+
+func TestConnIORecvLoopPanicAfterAuthorization(t *testing.T) {
+	t.Parallel()
+	ctx := testutil.Context(t, testutil.WaitShort)
+	logger := slogtest.Make(t, &slogtest.Options{IgnoreErrors: true})
+	bindings := make(chan binding, 1)
+	requests := make(chan *proto.CoordinateRequest, 1)
+	responses := make(chan *proto.CoordinateResponse, 1)
+	rfhs := make(chan readyForHandshake)
+	close(rfhs)
+	authorized := make(chan struct{}, 1)
+	peerID := uuid.New()
+	destinationID := uuid.New()
+	c := newConnIO(ctx, ctx, logger, bindings, make(chan tunnel, 1), rfhs,
+		requests, responses, peerID, t.Name(), agpltest.CoordinateeAuthFunc(func(context.Context, *proto.CoordinateRequest) error {
+			authorized <- struct{}{}
+			return nil
+		}))
+	t.Cleanup(func() { require.NoError(t, c.Close()) })
+	c.setLatestMapping([]mapping{{peer: destinationID}})
+
+	testutil.RequireSend(ctx, t, requests, &proto.CoordinateRequest{
+		ReadyForHandshake: []*proto.CoordinateRequest_ReadyForHandshake{{Id: agpl.UUIDToByteSlice(destinationID)}},
+	})
+	testutil.RequireReceive(ctx, t, authorized)
+	response := testutil.RequireReceive(ctx, t, responses)
+	require.Equal(t, agpl.CloseErrInternal, response.Error)
+	require.NotContains(t, response.Error, "send on closed channel")
+	select {
+	case _, ok := <-responses:
+		require.False(t, ok)
+	case <-ctx.Done():
+		t.Fatal("response channel did not close")
+	}
+	select {
+	case <-c.Done():
+	case <-ctx.Done():
+		t.Fatal("connection did not close")
+	}
+	withdrawn := testutil.RequireReceive(ctx, t, bindings)
+	require.Equal(t, bKey(peerID), withdrawn.bKey)
+	require.Equal(t, proto.CoordinateResponse_PeerUpdate_LOST, withdrawn.kind)
+}

enterprise/tailnet/mapper_internal_test.go

@@ -0,0 +1,55 @@
+package tailnet
+
+import (
+	"testing"
+
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/require"
+
+	"cdr.dev/slog/v3/sloggers/slogtest"
+	agpl "github.com/coder/coder/v2/tailnet"
+	"github.com/coder/coder/v2/tailnet/proto"
+	"github.com/coder/coder/v2/testutil"
+)
+
+func TestMapperPanic(t *testing.T) {
+	t.Parallel()
+	ctx := testutil.Context(t, testutil.WaitLong)
+	logger := slogtest.Make(t, &slogtest.Options{IgnoreErrors: true})
+	bindings := make(chan binding, 1)
+	responses := make(chan *proto.CoordinateResponse, 2)
+	c := newConnIO(ctx, ctx, logger, bindings, make(chan tunnel, 1), nil,
+		make(chan *proto.CoordinateRequest), responses, uuid.New(), t.Name(), agpl.SingleTailnetCoordinateeAuth{})
+	t.Cleanup(func() { require.NoError(t, c.Close()) })
+	coordinatorID := uuid.New()
+	m := &mapper{
+		ctx:        c.peerCtx,
+		logger:     logger,
+		c:          c,
+		mappings:   make(chan []mapping),
+		heartbeats: &heartbeats{self: coordinatorID},
+		// A nil sent map forces a panic after the mapping has been selected.
+		sent: nil,
+	}
+	go m.run()
+	require.NoError(t, agpl.SendCtx(ctx, m.mappings, []mapping{{
+		peer: uuid.New(), coordinator: coordinatorID,
+		node: &proto.Node{}, kind: proto.CoordinateResponse_PeerUpdate_NODE,
+	}}))
+	response := testutil.RequireReceive(ctx, t, responses)
+	require.Equal(t, agpl.CloseErrInternal, response.Error)
+	select {
+	case _, ok := <-responses:
+		require.False(t, ok)
+	case <-ctx.Done():
+		t.Fatal("response channel did not close")
+	}
+	select {
+	case <-c.Done():
+	case <-ctx.Done():
+		t.Fatal("connection did not close")
+	}
+	withdrawn := testutil.RequireReceive(ctx, t, bindings)
+	require.Equal(t, bKey(c.id), withdrawn.bKey)
+	require.Equal(t, proto.CoordinateResponse_PeerUpdate_LOST, withdrawn.kind)
+}

enterprise/tailnet/pgcoord.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"database/sql"
 	"math"
+	"runtime/debug"
 	"slices"
 	"strings"
 	"sync"
@@ -742,6 +743,16 @@ func newMapper(c *connIO, logger slog.Logger, h *heartbeats) *mapper {
 }
 
 func (m *mapper) run() {
+	defer func() {
+		if recovered := recover(); recovered != nil {
+			m.logger.Error(m.ctx, "panic processing peer mappings (recovered)",
+				slog.F("panic", recovered),
+				slog.F("stack", string(debug.Stack())),
+			)
+			_ = m.c.Enqueue(&proto.CoordinateResponse{Error: agpl.CloseErrInternal})
+			_ = m.c.Close()
+		}
+	}()
 	for {
 		var best map[uuid.UUID]mapping
 		select {

enterprise/tailnet/requests_test.go

@@ -0,0 +1,33 @@
+package tailnet_test
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"cdr.dev/slog/v3/sloggers/slogtest"
+	"github.com/coder/coder/v2/coderd/database/dbtestutil"
+	"github.com/coder/coder/v2/enterprise/tailnet"
+	"github.com/coder/coder/v2/tailnet/test"
+	"github.com/coder/coder/v2/testutil"
+)
+
+func TestPGCoordinator_InvalidRequests(t *testing.T) {
+	t.Parallel()
+	ctx := testutil.Context(t, testutil.WaitLong)
+	store, ps := dbtestutil.NewDB(t)
+	coordinator, err := tailnet.NewPGCoord(ctx, slogtest.Make(t, &slogtest.Options{IgnoreErrors: true}), ps, store)
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = coordinator.Close() })
+	test.InvalidCoordinateRequestTest(ctx, t, coordinator)
+}
+
+func TestPGCoordinator_RequestPanic(t *testing.T) {
+	t.Parallel()
+	ctx := testutil.Context(t, testutil.WaitLong)
+	store, ps := dbtestutil.NewDB(t)
+	coordinator, err := tailnet.NewPGCoord(ctx, slogtest.Make(t, &slogtest.Options{IgnoreErrors: true}), ps, store)
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = coordinator.Close() })
+	test.CoordinateRequestPanicTest(ctx, t, coordinator)
+}

tailnet/coordinator.go

@@ -7,6 +7,7 @@ import (
 	"io"
 	"net/http"
 	"net/netip"
+	runtimedebug "runtime/debug"
 	"sync"
 	"time"
 
@@ -27,6 +28,7 @@ const (
 	RequestBufferSize        = 32
 	CloseErrOverwritten      = "peer ID overwritten by new connection"
 	CloseErrCoordinatorClose = "coordinator closed"
+	CloseErrInternal         = "internal coordinator error"
 	ReadyForHandshakeError   = "ready for handshake error"
 )
 
@@ -176,6 +178,15 @@ func (c *coordinator) Coordinate(
 	c.wg.Add(1)
 	go func() {
 		defer c.wg.Done()
+		// Peer cleanup can also panic, outside reqLoop's recovery boundary.
+		defer func() {
+			if recovered := recover(); recovered != nil {
+				logger.Error(ctx, "panic coordinating peer (recovered)",
+					slog.F("panic", recovered),
+					slog.F("stack", string(runtimedebug.Stack())),
+				)
+			}
+		}()
 		loopErr := p.reqLoop(ctx, logger, c.core.handleRequest)
 		closeErrStr := ""
 		if loopErr != nil {
@@ -235,6 +246,9 @@ func (c *core) node(id uuid.UUID) *Node {
 }
 
 func (c *core) handleRequest(ctx context.Context, p *peer, req *proto.CoordinateRequest) error {
+	if err := ValidateCoordinateRequest(req); err != nil {
+		return err
+	}
 	c.mutex.Lock()
 	defer c.mutex.Unlock()
 	if c.closed {
@@ -290,6 +304,10 @@ func (c *core) handleRequest(ctx context.Context, p *peer, req *proto.Coordinate
 	}
 	if req.Disconnect != nil {
 		c.removePeerLocked(p.id, proto.CoordinateResponse_PeerUpdate_DISCONNECTED, "graceful disconnect", "")
+		// The peer and its response channel are gone, so nothing else in this
+		// request can be delivered. Matches the enterprise coordinator, which
+		// stops processing a request once it sees Disconnect.
+		return nil
 	}
 	if rfhs := req.ReadyForHandshake; rfhs != nil {
 		err := c.handleReadyForHandshakeLocked(pr, rfhs)
@@ -349,6 +367,10 @@ func (c *core) nodeUpdateLocked(p *peer, node *proto.Node) (err error) {
 
 	p.node = node
 	c.updateTunnelPeersLocked(p.id, node, proto.CoordinateResponse_PeerUpdate_NODE, "node update")
+	// Fan-out can recursively remove this peer if its response buffer is full.
+	if c.peers[p.id] != p {
+		return ErrAlreadyRemoved
+	}
 	return nil
 }
 
@@ -480,6 +502,13 @@ func (c *core) removePeerLocked(id uuid.UUID, kind proto.CoordinateResponse_Peer
 		return
 	}
 	c.updateTunnelPeersLocked(id, nil, kind, reason)
+	if _, ok := c.peers[id]; !ok {
+		// A tunnel peer with a full response buffer gets removed during the
+		// fan-out above, and removing it fans out to us in turn. When that
+		// nested call already closed and deleted this peer, p is stale and
+		// closing p.resps again would panic. Nothing is left to do.
+		return
+	}
 	c.tunnels.removeAll(id)
 	if closeErr != "" {
 		select {

tailnet/coordinator_internal_test.go

@@ -2,6 +2,7 @@ package tailnet
 
 import (
 	"bytes"
+	"context"
 	"flag"
 	"os"
 	"path/filepath"
@@ -11,6 +12,11 @@ import (
 
 	"github.com/google/uuid"
 	"github.com/stretchr/testify/require"
+
+	"cdr.dev/slog/v3"
+	"cdr.dev/slog/v3/sloggers/slogtest"
+	"github.com/coder/coder/v2/tailnet/proto"
+	"github.com/coder/coder/v2/testutil"
 )
 
 // UpdateGoldenFiles indicates golden files should be updated.
@@ -72,3 +78,209 @@ func TestDebugTemplate(t *testing.T) {
 		goldenPath,
 	)
 }
+
+// TestCoreDisconnectIgnoresReadyForHandshake sends Disconnect and
+// ReadyForHandshake in one request. Disconnect closes the peer's response
+// channel, so the handler must stop there instead of trying to answer the
+// handshake on a closed channel. Other peers still see DISCONNECTED.
+func TestCoreDisconnectIgnoresReadyForHandshake(t *testing.T) {
+	t.Parallel()
+	ctx := testutil.Context(t, testutil.WaitShort)
+	logger := testutil.Logger(t)
+	c := newCore(logger)
+	t.Cleanup(func() { require.NoError(t, c.close()) })
+
+	type testPeer struct {
+		*peer
+		resps chan *proto.CoordinateResponse
+	}
+	newPeer := func(name string, id uuid.UUID, auth CoordinateeAuth) testPeer {
+		resps := make(chan *proto.CoordinateResponse, ResponseBufferSize)
+		p := &peer{
+			logger: logger.With(slog.F("peer_name", name)),
+			id:     id,
+			name:   name,
+			resps:  resps,
+			reqs:   make(chan *proto.CoordinateRequest, RequestBufferSize),
+			auth:   auth,
+			sent:   make(map[uuid.UUID]*proto.Node),
+		}
+		require.NoError(t, c.initPeer(p))
+		return testPeer{peer: p, resps: resps}
+	}
+	observer := newPeer("observer", uuid.New(), SingleTailnetCoordinateeAuth{})
+	leaverID := uuid.New()
+	leaver := newPeer("leaver", leaverID, AgentCoordinateeAuth{ID: leaverID})
+
+	require.NoError(t, c.handleRequest(ctx, observer.peer, &proto.CoordinateRequest{
+		AddTunnel: &proto.CoordinateRequest_Tunnel{Id: leaver.id[:]},
+	}))
+	require.NoError(t, c.handleRequest(ctx, leaver.peer, &proto.CoordinateRequest{
+		UpdateSelf: &proto.CoordinateRequest_UpdateSelf{Node: &proto.Node{PreferredDerp: 1}},
+	}))
+	nodeUpdate := testutil.RequireReceive(ctx, t, observer.resps)
+	require.Len(t, nodeUpdate.PeerUpdates, 1)
+	require.Equal(t, proto.CoordinateResponse_PeerUpdate_NODE, nodeUpdate.PeerUpdates[0].Kind)
+
+	stranger := uuid.New()
+	var err error
+	require.NotPanics(t, func() {
+		err = c.handleRequest(ctx, leaver.peer, &proto.CoordinateRequest{
+			Disconnect:        &proto.CoordinateRequest_Disconnect{},
+			ReadyForHandshake: []*proto.CoordinateRequest_ReadyForHandshake{{Id: stranger[:]}},
+		})
+	})
+	require.NoError(t, err)
+
+	// The leaver's channel closes without an error response.
+	for closed := false; !closed; {
+		select {
+		case resp, ok := <-leaver.resps:
+			if !ok {
+				closed = true
+				continue
+			}
+			require.Empty(t, resp.Error)
+		case <-ctx.Done():
+			t.Fatal("leaver response channel was not closed")
+		}
+	}
+	c.mutex.RLock()
+	_, stillPresent := c.peers[leaver.id]
+	c.mutex.RUnlock()
+	require.False(t, stillPresent)
+
+	disconnected := testutil.RequireReceive(ctx, t, observer.resps)
+	require.Len(t, disconnected.PeerUpdates, 1)
+	gotID, err := uuid.FromBytes(disconnected.PeerUpdates[0].Id)
+	require.NoError(t, err)
+	require.Equal(t, leaver.id, gotID)
+	require.Equal(t, proto.CoordinateResponse_PeerUpdate_DISCONNECTED, disconnected.PeerUpdates[0].Kind)
+}
+
+// TestCoreRemovePeerNestedRemoval sets up two tunneled peers whose response
+// buffers are both full, then removes one of them. Notifying the other peer
+// fails with ErrWouldBlock, which removes that peer too, and notifying back
+// removes the first peer inside the nested call. The outer removePeerLocked
+// must notice its peer is already gone instead of closing the channel again.
+// Both request handling and lostPeer cleanup must finish without relying on
+// panic recovery.
+func TestCoreRemovePeerNestedRemoval(t *testing.T) {
+	t.Parallel()
+
+	type fullPeers struct {
+		core   *core
+		a, b   *peer
+		aResps chan *proto.CoordinateResponse
+		bResps chan *proto.CoordinateResponse
+	}
+	// setup returns a core with peers a and b sharing a tunnel, where each
+	// peer's single-slot response buffer already holds the other's node.
+	setup := func(t *testing.T) fullPeers {
+		// The full buffers make updateTunnelPeersLocked log "failed to update
+		// mapping" at Error, which is the expected trigger. Anything else at
+		// Error or Critical, such as "removed non-existent peer", still fails
+		// the test.
+		logger := slogtest.Make(t, &slogtest.Options{
+			IgnoreErrorFn: func(ent slog.SinkEntry) bool {
+				return ent.Message == "failed to update mapping"
+			},
+		}).Leveled(slog.LevelDebug)
+		c := newCore(logger)
+		t.Cleanup(func() { require.NoError(t, c.close()) })
+		newPeer := func(name string) (*peer, chan *proto.CoordinateResponse) {
+			resps := make(chan *proto.CoordinateResponse, 1)
+			p := &peer{
+				logger: logger.With(slog.F("peer_name", name)),
+				id:     uuid.New(),
+				name:   name,
+				resps:  resps,
+				reqs:   make(chan *proto.CoordinateRequest, RequestBufferSize),
+				auth:   SingleTailnetCoordinateeAuth{},
+				sent:   make(map[uuid.UUID]*proto.Node),
+			}
+			require.NoError(t, c.initPeer(p))
+			return p, resps
+		}
+		a, aResps := newPeer("a")
+		b, bResps := newPeer("b")
+
+		c.mutex.Lock()
+		defer c.mutex.Unlock()
+		// No tunnel yet, so these updates are not sent anywhere.
+		require.NoError(t, c.nodeUpdateLocked(a, &proto.Node{PreferredDerp: 1}))
+		require.NoError(t, c.nodeUpdateLocked(b, &proto.Node{PreferredDerp: 2}))
+		// Adding the tunnel sends each node to the other peer, which fills
+		// both buffers.
+		require.NoError(t, c.addTunnelLocked(a, b.id))
+		require.Len(t, aResps, 1)
+		require.Len(t, bResps, 1)
+		require.Contains(t, a.sent, b.id)
+		require.Contains(t, b.sent, a.id)
+		return fullPeers{core: c, a: a, b: b, aResps: aResps, bResps: bResps}
+	}
+
+	requireClosed := func(ctx context.Context, t *testing.T, ch chan *proto.CoordinateResponse) {
+		for {
+			select {
+			case _, ok := <-ch:
+				if !ok {
+					return
+				}
+			case <-ctx.Done():
+				t.Fatal("response channel was not closed")
+			}
+		}
+	}
+
+	requireBothRemoved := func(ctx context.Context, t *testing.T, fp fullPeers) {
+		fp.core.mutex.RLock()
+		_, aPresent := fp.core.peers[fp.a.id]
+		_, bPresent := fp.core.peers[fp.b.id]
+		aTunnels := fp.core.tunnels.findTunnelPeers(fp.a.id)
+		bTunnels := fp.core.tunnels.findTunnelPeers(fp.b.id)
+		fp.core.mutex.RUnlock()
+		require.False(t, aPresent)
+		require.False(t, bPresent)
+		require.Empty(t, aTunnels)
+		require.Empty(t, bTunnels)
+		requireClosed(ctx, t, fp.aResps)
+		requireClosed(ctx, t, fp.bResps)
+	}
+
+	for _, name := range []string{"UpdateSelf", "UpdateSelfAddTunnel", "UpdateSelfReadyForHandshake"} {
+		t.Run(name, func(t *testing.T) {
+			t.Parallel()
+			ctx := testutil.Context(t, testutil.WaitShort)
+			fp := setup(t)
+			req := &proto.CoordinateRequest{
+				UpdateSelf: &proto.CoordinateRequest_UpdateSelf{Node: &proto.Node{PreferredDerp: 3}},
+			}
+			dstID := uuid.New()
+			switch name {
+			case "UpdateSelfAddTunnel":
+				req.AddTunnel = &proto.CoordinateRequest_Tunnel{Id: dstID[:]}
+			case "UpdateSelfReadyForHandshake":
+				req.ReadyForHandshake = []*proto.CoordinateRequest_ReadyForHandshake{{Id: dstID[:]}}
+			}
+			var err error
+			require.NotPanics(t, func() {
+				err = fp.core.handleRequest(ctx, fp.a, req)
+			})
+			require.NoError(t, err)
+			requireBothRemoved(ctx, t, fp)
+		})
+	}
+
+	t.Run("LostPeer", func(t *testing.T) {
+		t.Parallel()
+		ctx := testutil.Context(t, testutil.WaitShort)
+		fp := setup(t)
+		var err error
+		require.NotPanics(t, func() {
+			err = fp.core.lostPeer(fp.a, "")
+		})
+		require.NoError(t, err)
+		requireBothRemoved(ctx, t, fp)
+	})
+}

tailnet/peer.go

@@ -2,6 +2,7 @@ package tailnet
 
 import (
 	"context"
+	"runtime/debug"
 	"time"
 
 	"github.com/google/uuid"
@@ -121,7 +122,16 @@ func (p *peer) storeMappingLocked(
 	}, nil
 }
 
-func (p *peer) reqLoop(ctx context.Context, logger slog.Logger, handler func(context.Context, *peer, *proto.CoordinateRequest) error) error {
+func (p *peer) reqLoop(ctx context.Context, logger slog.Logger, handler func(context.Context, *peer, *proto.CoordinateRequest) error) (err error) {
+	defer func() {
+		if recovered := recover(); recovered != nil {
+			logger.Error(ctx, "peerReadLoop panic handling request (recovered)",
+				slog.F("panic", recovered),
+				slog.F("stack", string(debug.Stack())),
+			)
+			err = xerrors.New(CloseErrInternal)
+		}
+	}()
 	for {
 		select {
 		case <-ctx.Done():

tailnet/peer_internal_test.go

@@ -0,0 +1,119 @@
+package tailnet
+
+import (
+	"context"
+	"testing"
+
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/require"
+
+	"cdr.dev/slog/v3"
+	"cdr.dev/slog/v3/sloggers/slogtest"
+	"github.com/coder/coder/v2/tailnet/proto"
+	"github.com/coder/coder/v2/testutil"
+)
+
+func TestPeerReqLoopPanic(t *testing.T) {
+	t.Parallel()
+	ctx := testutil.Context(t, testutil.WaitShort)
+	logger := slogtest.Make(t, &slogtest.Options{IgnoreErrors: true})
+	requests := make(chan *proto.CoordinateRequest, 1)
+	responses := make(chan *proto.CoordinateResponse, 1)
+	p := &peer{
+		logger: logger,
+		id:     uuid.New(),
+		resps:  responses,
+		reqs:   requests,
+		sent:   make(map[uuid.UUID]*proto.Node),
+	}
+	core := newCore(logger)
+	require.NoError(t, core.initPeer(p))
+	t.Cleanup(func() { require.NoError(t, core.close()) })
+
+	testutil.RequireSend(ctx, t, requests, &proto.CoordinateRequest{})
+	err := p.reqLoop(ctx, logger, func(context.Context, *peer, *proto.CoordinateRequest) error {
+		panic("private handler panic")
+	})
+	require.EqualError(t, err, CloseErrInternal)
+	require.NoError(t, core.lostPeer(p, err.Error()))
+
+	response := testutil.RequireReceive(ctx, t, responses)
+	require.Equal(t, CloseErrInternal, response.Error)
+	require.NotContains(t, response.Error, "private handler panic")
+	_, ok := <-responses
+	require.False(t, ok)
+	core.mutex.RLock()
+	_, ok = core.peers[p.id]
+	core.mutex.RUnlock()
+	require.False(t, ok)
+}
+
+func TestPeerCleanupPanic(t *testing.T) {
+	t.Parallel()
+	ctx := testutil.Context(t, testutil.WaitLong)
+	id := uuid.New()
+	sink := &peerCleanupPanicSink{
+		FakeSink: testutil.NewFakeSink(t),
+		peerID:   id,
+	}
+	logger := slog.Make(sink).Leveled(slog.LevelDebug)
+	c := NewCoordinator(logger).(*coordinator)
+	t.Cleanup(func() { require.NoError(t, c.Close()) })
+
+	requests, responses := c.Coordinate(ctx, id, t.Name(), SingleTailnetCoordinateeAuth{})
+	close(requests)
+	done := make(chan struct{})
+	go func() {
+		c.wg.Wait()
+		close(done)
+	}()
+	testutil.TryReceive(ctx, t, done)
+
+	entries := sink.Entries(func(entry slog.SinkEntry) bool {
+		return entry.Level == slog.LevelError
+	})
+	require.Len(t, entries, 1)
+	require.Equal(t, "panic coordinating peer (recovered)", entries[0].Message)
+	fields := make(map[string]any)
+	for _, field := range entries[0].Fields {
+		fields[field.Name] = field.Value
+	}
+	require.Equal(t, id, fields["peer_id"])
+	require.Equal(t, "private cleanup panic", fields["panic"])
+	require.Contains(t, fields["stack"], "(*core).lostPeer")
+
+	// Recovery must leave the core unlocked and the coordinator usable.
+	healthyID := uuid.New()
+	healthyRequests, healthyResponses := c.Coordinate(ctx, healthyID, t.Name(), SingleTailnetCoordinateeAuth{})
+	close(healthyRequests)
+	require.Nil(t, testutil.TryReceive(ctx, t, healthyResponses))
+
+	// Shutdown must still remove the peer whose cleanup was interrupted.
+	closed := make(chan error, 1)
+	go func() { closed <- c.Close() }()
+	require.NoError(t, testutil.RequireReceive(ctx, t, closed))
+	response := testutil.RequireReceive(ctx, t, responses)
+	require.Equal(t, CloseErrCoordinatorClose, response.Error)
+	require.NotContains(t, response.Error, "private cleanup panic")
+	require.Nil(t, testutil.TryReceive(ctx, t, responses))
+	c.core.mutex.RLock()
+	_, present := c.core.peers[id]
+	c.core.mutex.RUnlock()
+	require.False(t, present)
+}
+
+type peerCleanupPanicSink struct {
+	*testutil.FakeSink
+	peerID uuid.UUID
+}
+
+func (s *peerCleanupPanicSink) LogEntry(ctx context.Context, entry slog.SinkEntry) {
+	if entry.Message == "lostPeer" {
+		for _, field := range entry.Fields {
+			if field.Name == "peer_id" && field.Value == s.peerID {
+				panic("private cleanup panic")
+			}
+		}
+	}
+	s.FakeSink.LogEntry(ctx, entry)
+}

tailnet/requests.go

@@ -0,0 +1,31 @@
+package tailnet
+
+import (
+	"slices"
+
+	"golang.org/x/xerrors"
+
+	"github.com/coder/coder/v2/tailnet/proto"
+)
+
+// ValidateCoordinateRequest checks the structural invariants required by the
+// coordinator before authorization or state updates.
+func ValidateCoordinateRequest(req *proto.CoordinateRequest) error {
+	if req == nil {
+		return xerrors.New("coordinate request is required")
+	}
+	if err := validateUpdateSelf(req); err != nil {
+		return err
+	}
+	if slices.Contains(req.ReadyForHandshake, nil) {
+		return xerrors.New("ready_for_handshake entries must not be nil")
+	}
+	return nil
+}
+
+func validateUpdateSelf(req *proto.CoordinateRequest) error {
+	if upd := req.GetUpdateSelf(); upd != nil && upd.GetNode() == nil {
+		return xerrors.New("update_self node is required")
+	}
+	return nil
+}

tailnet/requests_test.go

@@ -0,0 +1,100 @@
+package tailnet_test
+
+import (
+	"testing"
+
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/require"
+
+	"cdr.dev/slog/v3/sloggers/slogtest"
+	"github.com/coder/coder/v2/tailnet"
+	"github.com/coder/coder/v2/tailnet/proto"
+	"github.com/coder/coder/v2/tailnet/test"
+	"github.com/coder/coder/v2/testutil"
+)
+
+func TestCoordinateeAuthNilNode(t *testing.T) {
+	t.Parallel()
+	for name, auth := range map[string]tailnet.CoordinateeAuth{
+		"Agent":      tailnet.AgentCoordinateeAuth{ID: uuid.New()},
+		"Client":     tailnet.ClientCoordinateeAuth{AgentID: uuid.New()},
+		"ClientUser": tailnet.ClientUserCoordinateeAuth{},
+	} {
+		t.Run(name, func(t *testing.T) {
+			t.Parallel()
+			ctx := testutil.Context(t, testutil.WaitShort)
+			require.NotPanics(t, func() {
+				require.EqualError(t, auth.Authorize(ctx, &proto.CoordinateRequest{
+					UpdateSelf: &proto.CoordinateRequest_UpdateSelf{},
+				}), "update_self node is required")
+			})
+		})
+	}
+}
+
+func TestCoordinator_InvalidRequests(t *testing.T) {
+	t.Parallel()
+	ctx := testutil.Context(t, testutil.WaitLong)
+	coordinator := tailnet.NewCoordinator(slogtest.Make(t, &slogtest.Options{IgnoreErrors: true}))
+	t.Cleanup(func() { require.NoError(t, coordinator.Close()) })
+	test.InvalidCoordinateRequestTest(ctx, t, coordinator)
+}
+
+func TestCoordinator_RequestPanic(t *testing.T) {
+	t.Parallel()
+	ctx := testutil.Context(t, testutil.WaitLong)
+	coordinator := tailnet.NewCoordinator(slogtest.Make(t, &slogtest.Options{IgnoreErrors: true}))
+	t.Cleanup(func() { require.NoError(t, coordinator.Close()) })
+	test.CoordinateRequestPanicTest(ctx, t, coordinator)
+}
+
+func TestValidateCoordinateRequest(t *testing.T) {
+	t.Parallel()
+	validRFH := &proto.CoordinateRequest_ReadyForHandshake{Id: tailnet.UUIDToByteSlice(uuid.New())}
+	for _, tc := range []struct {
+		name string
+		req  *proto.CoordinateRequest
+		err  string // empty means the request is accepted
+	}{
+		{name: "NilRequest", err: "coordinate request is required"},
+		{name: "EmptyRequest", req: &proto.CoordinateRequest{}},
+		{
+			name: "NilNode",
+			req:  &proto.CoordinateRequest{UpdateSelf: &proto.CoordinateRequest_UpdateSelf{}},
+			err:  "update_self node is required",
+		},
+		{
+			name: "EmptyNode",
+			req:  &proto.CoordinateRequest{UpdateSelf: &proto.CoordinateRequest_UpdateSelf{Node: &proto.Node{}}},
+		},
+		{name: "NilReadyForHandshake", req: &proto.CoordinateRequest{ReadyForHandshake: nil}},
+		{
+			name: "EmptyReadyForHandshake",
+			req:  &proto.CoordinateRequest{ReadyForHandshake: []*proto.CoordinateRequest_ReadyForHandshake{}},
+		},
+		{
+			name: "ValidReadyForHandshake",
+			req:  &proto.CoordinateRequest{ReadyForHandshake: []*proto.CoordinateRequest_ReadyForHandshake{validRFH}},
+		},
+		{
+			name: "NilReadyForHandshakeEntry",
+			req:  &proto.CoordinateRequest{ReadyForHandshake: []*proto.CoordinateRequest_ReadyForHandshake{nil}},
+			err:  "ready_for_handshake entries must not be nil",
+		},
+		{
+			name: "NilReadyForHandshakeAfterValid",
+			req:  &proto.CoordinateRequest{ReadyForHandshake: []*proto.CoordinateRequest_ReadyForHandshake{validRFH, nil}},
+			err:  "ready_for_handshake entries must not be nil",
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			err := tailnet.ValidateCoordinateRequest(tc.req)
+			if tc.err == "" {
+				require.NoError(t, err)
+				return
+			}
+			require.EqualError(t, err, tc.err)
+		})
+	}
+}

tailnet/test/peer.go

@@ -407,3 +407,12 @@ func (f FakeCoordinateeAuth) Authorize(ctx context.Context, _ *proto.CoordinateR
 }
 
 var _ tailnet.CoordinateeAuth = (*FakeCoordinateeAuth)(nil)
+
+// CoordinateeAuthFunc adapts a plain function to tailnet.CoordinateeAuth so a
+// test can return a specific error or panic on a chosen request without
+// declaring a new type.
+type CoordinateeAuthFunc func(context.Context, *proto.CoordinateRequest) error
+
+func (f CoordinateeAuthFunc) Authorize(ctx context.Context, req *proto.CoordinateRequest) error {
+	return f(ctx, req)
+}

tailnet/test/requests.go

@@ -0,0 +1,116 @@
+package test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/coder/coder/v2/tailnet"
+	"github.com/coder/coder/v2/tailnet/proto"
+)
+
+// InvalidCoordinateRequestTest checks that malformed requests are rejected
+// before authorization and that a valid node bundled with a malformed entry
+// is not published to tunnel peers.
+func InvalidCoordinateRequestTest(ctx context.Context, t *testing.T, coordinator tailnet.CoordinatorV2) {
+	const (
+		baselineDERP  = 31001
+		malformedDERP = 31002
+	)
+	for _, tc := range []struct {
+		name             string
+		req              *proto.CoordinateRequest
+		err              string
+		verifyNoMutation bool
+	}{
+		{name: "NilRequest", err: "coordinate request is required"},
+		{
+			name: "NilNode",
+			req:  &proto.CoordinateRequest{UpdateSelf: &proto.CoordinateRequest_UpdateSelf{}},
+			err:  "update_self node is required",
+		},
+		{
+			name: "NilReadyForHandshake",
+			req:  &proto.CoordinateRequest{ReadyForHandshake: []*proto.CoordinateRequest_ReadyForHandshake{nil}},
+			err:  "ready_for_handshake entries must not be nil",
+		},
+		{
+			name: "NodeAndNilReadyForHandshake",
+			req: &proto.CoordinateRequest{
+				UpdateSelf: &proto.CoordinateRequest_UpdateSelf{Node: &proto.Node{
+					PreferredDerp: malformedDERP,
+				}},
+				ReadyForHandshake: []*proto.CoordinateRequest_ReadyForHandshake{nil},
+			},
+			err:              "ready_for_handshake entries must not be nil",
+			verifyNoMutation: true,
+		},
+	} {
+		// The cases share the caller's ctx and coordinator, so they run
+		// sequentially. Parallel subtests would pause until the caller
+		// returns and, under -parallel=1, could resume after that ctx has
+		// already expired.
+		t.Run(tc.name, func(t *testing.T) {
+			var auth tailnet.CoordinateeAuth = CoordinateeAuthFunc(func(context.Context, *proto.CoordinateRequest) error {
+				panic("authorization must not see malformed requests")
+			})
+			if tc.verifyNoMutation {
+				auth = tailnet.SingleTailnetCoordinateeAuth{}
+			}
+			peer := NewPeer(ctx, t, coordinator, tc.name, WithAuth(auth))
+			defer peer.Close(ctx)
+			var observer *Peer
+			if tc.verifyNoMutation {
+				observer = NewPeer(ctx, t, coordinator, tc.name+"Observer")
+				defer observer.Close(ctx)
+				observer.AddTunnel(peer.ID)
+				peer.UpdateDERP(baselineDERP)
+				observer.AssertEventuallyHasDERP(peer.ID, baselineDERP)
+			}
+			require.NoError(t, tailnet.SendCtx(ctx, peer.reqs, tc.req))
+			peer.AssertEventuallyResponsesClosed(tc.err)
+			if tc.verifyNoMutation {
+				observer.AssertEventuallyLost(peer.ID)
+				for _, update := range observer.peerUpdates[peer.ID] {
+					require.NotEqual(t, int32(malformedDERP), update.GetNode().GetPreferredDerp())
+				}
+			}
+		})
+	}
+}
+
+// CoordinateRequestPanicTest checks that a failed connection is withdrawn while
+// other connections continue exchanging updates.
+func CoordinateRequestPanicTest(ctx context.Context, t *testing.T, coordinator tailnet.CoordinatorV2) {
+	broken := NewPeer(ctx, t, coordinator, "broken", WithAuth(CoordinateeAuthFunc(func(_ context.Context, req *proto.CoordinateRequest) error {
+		if req.Disconnect != nil {
+			panic("private panic detail")
+		}
+		return nil
+	})))
+	defer broken.Close(ctx)
+	healthy := NewPeer(ctx, t, coordinator, "healthy")
+	defer healthy.Close(ctx)
+	other := NewPeer(ctx, t, coordinator, "other")
+	defer other.Close(ctx)
+	healthy.AddTunnel(broken.ID)
+	healthy.AddTunnel(other.ID)
+	broken.UpdateDERP(1)
+	healthy.UpdateDERP(2)
+	other.UpdateDERP(3)
+	healthy.AssertEventuallyHasDERP(broken.ID, 1)
+	healthy.AssertEventuallyHasDERP(other.ID, 3)
+
+	broken.Disconnect()
+	broken.AssertEventuallyResponsesClosed(tailnet.CloseErrInternal)
+	healthy.AssertEventuallyLost(broken.ID)
+
+	// Empty messages are valid no-ops, and an empty Node is still a Node.
+	require.NoError(t, tailnet.SendCtx(ctx, other.reqs, &proto.CoordinateRequest{}))
+	require.NoError(t, tailnet.SendCtx(ctx, other.reqs, &proto.CoordinateRequest{
+		UpdateSelf: &proto.CoordinateRequest_UpdateSelf{Node: &proto.Node{}},
+	}))
+	other.UpdateDERP(4)
+	healthy.AssertEventuallyHasDERP(other.ID, 4)
+}

tailnet/tunnel.go

@@ -66,6 +66,9 @@ type AgentCoordinateeAuth struct {
 }
 
 func (a AgentCoordinateeAuth) Authorize(_ context.Context, req *proto.CoordinateRequest) error {
+	if err := validateUpdateSelf(req); err != nil {
+		return err
+	}
 	if tun := req.GetAddTunnel(); tun != nil {
 		return xerrors.New("agents cannot open tunnels")
 	}
@@ -130,6 +133,9 @@ func (a ClientUserCoordinateeAuth) Authorize(ctx context.Context, req *proto.Coo
 
 // handleClientNodeRequests validates GetUpdateSelf requests and declines ReadyForHandshake requests
 func handleClientNodeRequests(req *proto.CoordinateRequest) error {
+	if err := validateUpdateSelf(req); err != nil {
+		return err
+	}
 	if upd := req.GetUpdateSelf(); upd != nil {
 		for _, addrStr := range upd.Node.Addresses {
 			pre, err := netip.ParsePrefix(addrStr)