diff --git a/enterprise/tailnet/connio.go b/enterprise/tailnet/connio.go index 7c186dc1a0..0803aa79f5 100644 --- a/enterprise/tailnet/connio.go +++ b/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)) diff --git a/enterprise/tailnet/connio_internal_test.go b/enterprise/tailnet/connio_internal_test.go new file mode 100644 index 0000000000..77649c2ed8 --- /dev/null +++ b/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) +} diff --git a/enterprise/tailnet/mapper_internal_test.go b/enterprise/tailnet/mapper_internal_test.go new file mode 100644 index 0000000000..f5c2c70f56 --- /dev/null +++ b/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) +} diff --git a/enterprise/tailnet/pgcoord.go b/enterprise/tailnet/pgcoord.go index 08325e567f..4018b8aabf 100644 --- a/enterprise/tailnet/pgcoord.go +++ b/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 { diff --git a/enterprise/tailnet/requests_test.go b/enterprise/tailnet/requests_test.go new file mode 100644 index 0000000000..32ebeeb2ad --- /dev/null +++ b/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) +} diff --git a/tailnet/coordinator.go b/tailnet/coordinator.go index c3df906b4d..79db8ca0a2 100644 --- a/tailnet/coordinator.go +++ b/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 { diff --git a/tailnet/coordinator_internal_test.go b/tailnet/coordinator_internal_test.go index 9f5ac7c6a4..654bf28015 100644 --- a/tailnet/coordinator_internal_test.go +++ b/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) + }) +} diff --git a/tailnet/peer.go b/tailnet/peer.go index 1954534a1d..8631c24e62 100644 --- a/tailnet/peer.go +++ b/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(): diff --git a/tailnet/peer_internal_test.go b/tailnet/peer_internal_test.go new file mode 100644 index 0000000000..1dd2083e82 --- /dev/null +++ b/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) +} diff --git a/tailnet/requests.go b/tailnet/requests.go new file mode 100644 index 0000000000..94cebe151e --- /dev/null +++ b/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 +} diff --git a/tailnet/requests_test.go b/tailnet/requests_test.go new file mode 100644 index 0000000000..8d50a50771 --- /dev/null +++ b/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) + }) + } +} diff --git a/tailnet/test/peer.go b/tailnet/test/peer.go index 601dc748d4..9dc44f431e 100644 --- a/tailnet/test/peer.go +++ b/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) +} diff --git a/tailnet/test/requests.go b/tailnet/test/requests.go new file mode 100644 index 0000000000..da84432604 --- /dev/null +++ b/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) +} diff --git a/tailnet/tunnel.go b/tailnet/tunnel.go index 23720d7d5c..9ea7d440c2 100644 --- a/tailnet/tunnel.go +++ b/tailnet/tunnel.go @@ -85,6 +85,9 @@ type AgentCoordinateeAuth struct { } func (a AgentCoordinateeAuth) Authorize(_ context.Context, req *proto.CoordinateRequest) error { + if err := validateUpdateSelf(req); err != nil { + return err + } if req.GetAddTunnel() != nil { return xerrors.New("agents cannot open tunnels") } @@ -161,6 +164,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)