[client] Bound the SSH session request queue and keep resizes from blocking the session loop (#8363)

* Bound the SSH session request queue by bytes and stop resizes reaching a loop with no consumer

* Give the window change drain loop a body

* Avoid an empty case clause in the SSH request filter

* Charge only the newest resize to the request queue and classify requests before the byte limit

client/ssh/server/session_handlers.go

@@ -53,13 +53,20 @@ func (s *Server) sessionHandler(session ssh.Session) {
 		logger.Infof("SSH session closed after %v", duration)
 	}()
 
+	ptyReq, winCh, isPty := session.Pty()
+	if isPty {
+		// The request loop blocks sending a resize nobody reads, and only sees its request
+		// stream close once that send completes. Covers the paths below that return
+		// without consuming window changes.
+		defer drainWindowChanges(winCh)
+	}
+
 	privilegeResult, err := s.userPrivilegeCheck(session.User())
 	if err != nil {
 		s.handlePrivError(logger, session, err)
 		return
 	}
 
-	ptyReq, winCh, isPty := session.Pty()
 	hasCommand := session.RawCommand() != ""
 
 	if isPty && !hasCommand {

client/ssh/server/session_hardening_test.go

@@ -338,3 +338,65 @@ func TestFilterSessionRequests_RefusedRequestsStayInOrder(t *testing.T) {
 		t.Fatal("filter did not close its output")
 	}
 }
+
+// A rejected subsystem starts no command, so nothing drains the window channel. A resize
+// after it must be refused instead of blocking the session's request loop.
+func TestSession_ResizeAfterRejectedSubsystemDoesNotStallLoop(t *testing.T) {
+	addr, hostKey := startNoAuthServer(t)
+	client := dialNoAuthServer(t, addr, hostKey)
+
+	session, err := client.NewSession()
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = session.Close() })
+
+	require.NoError(t, session.RequestPty("xterm", 24, 80, cryptossh.TerminalModes{}))
+	ok, err := session.SendRequest("subsystem", true, cryptossh.Marshal(struct{ Name string }{"nope"}))
+	require.NoError(t, err)
+	require.False(t, ok, "unknown subsystem must be rejected")
+
+	winch := cryptossh.Marshal(struct{ Columns, Rows, Width, Height uint32 }{100, 40, 0, 0})
+	_, err = session.SendRequest("window-change", false, winch)
+	require.NoError(t, err)
+
+	done := make(chan bool, 1)
+	go func() {
+		ok, err := session.SendRequest("env", true, cryptossh.Marshal(struct{ Key, Value string }{"A", "b"}))
+		done <- err == nil && ok
+	}()
+	select {
+	case ok := <-done:
+		assert.True(t, ok, "env request after the resize should be accepted")
+	case <-time.After(5 * time.Second):
+		t.Fatal("session request loop stalled behind a resize with no command running")
+	}
+}
+
+// A resize that reaches the loop while the command handler never reads window changes
+// (sftp) must not leave the loop goroutine behind once the handler returns.
+func TestSession_ResizeDuringSFTPDoesNotLeakRequestLoop(t *testing.T) {
+	addr, hostKey := startNoAuthServer(t)
+	loops := func() int {
+		buf := make([]byte, 1<<22)
+		return strings.Count(string(buf[:runtime.Stack(buf, true)]), "ssh.(*session).handleRequests")
+	}
+	before := loops()
+
+	winch := cryptossh.Marshal(struct{ Columns, Rows, Width, Height uint32 }{100, 40, 0, 0})
+	for i := 0; i < 10; i++ {
+		client := dialNoAuthServer(t, addr, hostKey)
+		session, err := client.NewSession()
+		require.NoError(t, err)
+		require.NoError(t, session.RequestPty("xterm", 24, 80, cryptossh.TerminalModes{}))
+		ok, err := session.SendRequest("subsystem", true, cryptossh.Marshal(struct{ Name string }{"sftp"}))
+		require.NoError(t, err)
+		require.True(t, ok, "the sftp subsystem must be accepted for the resize to matter")
+		// The sftp handler may already have ended the session, so a send can fail.
+		_, _ = session.SendRequest("window-change", false, winch)
+		_, _ = session.SendRequest("window-change", false, winch)
+		_ = session.Close()
+		_ = client.Close()
+	}
+
+	assert.Eventually(t, func() bool { return loops() <= before }, 10*time.Second, 100*time.Millisecond,
+		"session request loops must exit after the clients disconnect")
+}

client/ssh/server/session_limits.go

@@ -23,9 +23,11 @@ const (
 	maxEnvRequests = 128
 	maxEnvPayload  = 8 * 1024
 
-	// maxQueuedRequests bounds the requests held for a session whose request loop is
-	// not consuming them. A client that exceeds it has its channel closed.
-	maxQueuedRequests = 256
+	// maxQueuedRequests and maxQueuedBytes bound the requests held for a session whose
+	// request loop is not consuming them. A client that exceeds either has its channel
+	// closed.
+	maxQueuedRequests = 128
+	maxQueuedBytes    = 256 * 1024
 )
 
 type (
@@ -59,6 +61,18 @@ func (s *Server) sessionChannelHandler(srv *ssh.Server, conn *cryptossh.ServerCo
 	ssh.DefaultSessionHandler(srv, conn, guarded, ctx)
 }
 
+// drainWindowChanges discards window changes until winCh is closed. It must run after
+// the session's command handler has returned.
+func drainWindowChanges(winCh <-chan ssh.Window) {
+	go func() {
+		for {
+			if _, ok := <-winCh; !ok {
+				return
+			}
+		}
+	}()
+}
+
 // acquireConnSlot takes one of limit slots counted under key for the connection that
 // ctx belongs to. The returned func releases the slot.
 func acquireConnSlot(ctx ssh.Context, key any, limit int32) (func(), bool) {
@@ -114,16 +128,31 @@ const refusedRequestType = "refused@netbird.io"
 // what the loop can accumulate or block on bounded:
 //   - a malformed pty-req is refused instead of reaching the loop's parser
 //   - env requests beyond maxEnvRequests, or with an oversized payload, are refused
-//   - window-change requests are refused until the command starts and coalesced
-//     after, as the loop blocks sending on a one-slot channel that only the command
-//     handler drains
-//   - the queue behind a stalled loop is capped, after which the channel is closed
+//   - window-change requests are refused unless the session's command is a shell or
+//     exec, and coalesced after it starts, as the loop blocks sending on a one-slot
+//     channel that only a shell or exec handler drains. Subsystems never read it.
+//   - request types the loop does not act on are refused, which also drops their payload
+//   - the queue behind a stalled loop is capped by count and by payload bytes, after
+//     which the channel is closed
 type requestFilter struct {
-	ch        cryptossh.Channel
-	queue     []*cryptossh.Request
-	lastWinch *cryptossh.Request
-	started   bool
-	envCount  int
+	ch          cryptossh.Channel
+	queue       []*queuedRequest
+	queuedBytes int
+	lastWinch   *queuedRequest
+	envCount    int
+
+	// commandSeen is set by the first shell, exec or subsystem request, which decides
+	// the session's command as far as the loop is concerned.
+	commandSeen bool
+	// started is set when that first command is a shell or exec.
+	started bool
+}
+
+// queuedRequest remembers the payload size counted against the queue, as a refusal
+// drops the payload and releases that size.
+type queuedRequest struct {
+	req  *cryptossh.Request
+	size int
 }
 
 // filterSessionRequests returns the requests of in as the session loop should see them.
@@ -145,7 +174,7 @@ func (f *requestFilter) run(in <-chan *cryptossh.Request, out chan<- *cryptossh.
 		)
 		if len(f.queue) > 0 {
 			sendCh = out
-			next = f.queue[0]
+			next = f.queue[0].req
 		}
 
 		select {
@@ -163,49 +192,83 @@ func (f *requestFilter) run(in <-chan *cryptossh.Request, out chan<- *cryptossh.
 }
 
 func (f *requestFilter) dropHead() {
-	if f.queue[0] == f.lastWinch {
+	head := f.queue[0]
+	if head == f.lastWinch {
 		f.lastWinch = nil
 	}
+	f.queuedBytes -= head.size
+	f.queue[0] = nil
 	f.queue = f.queue[1:]
 }
 
 // enqueue classifies req and queues it. It returns false after closing the channel
-// because the queue is full.
+// because the queue is full. Requests are classified first so a refused request's
+// dropped payload does not count against the limit.
 func (f *requestFilter) enqueue(req *cryptossh.Request) bool {
-	if len(f.queue) >= maxQueuedRequests {
+	q := &queuedRequest{req: req}
+	f.classify(q)
+	q.size = len(req.Payload)
+
+	if len(f.queue) >= maxQueuedRequests || f.queuedBytes+q.size > maxQueuedBytes {
 		log.Warnf("SSH session request queue full, closing channel")
-		_ = f.ch.Close()
+		if f.ch != nil {
+			_ = f.ch.Close()
+		}
 		return false
 	}
 
+	f.queue = append(f.queue, q)
+	f.queuedBytes += q.size
+	return true
+}
+
+func (f *requestFilter) classify(q *queuedRequest) {
+	req := q.req
 	switch req.Type {
 	case "pty-req":
 		if !validPtyReqPayload(req.Payload) {
-			refuseRequest(req)
+			f.refuse(q)
 		}
 	case "env":
 		f.envCount++
 		if f.envCount > maxEnvRequests || len(req.Payload) > maxEnvPayload {
-			refuseRequest(req)
+			f.refuse(q)
 		}
 	case "shell", "exec", "subsystem":
-		f.started = true
+		f.noteCommand(req.Type)
 	case "window-change":
-		f.coalesceWindowChange(req)
+		f.coalesceWindowChange(q)
+	default:
+		if !loopActsOn(req.Type) {
+			f.refuse(q)
+		}
 	}
-	f.queue = append(f.queue, req)
-	return true
 }
 
-func (f *requestFilter) coalesceWindowChange(req *cryptossh.Request) {
-	if !f.started || !validWinchPayload(req.Payload) {
-		refuseRequest(req)
+// refuse drops the payload of q and releases the size it was charged against the queue.
+func (f *requestFilter) refuse(q *queuedRequest) {
+	f.queuedBytes -= q.size
+	q.size = 0
+	refuseRequest(q.req)
+}
+
+func (f *requestFilter) noteCommand(reqType string) {
+	if f.commandSeen {
+		return
+	}
+	f.commandSeen = true
+	f.started = reqType != "subsystem"
+}
+
+func (f *requestFilter) coalesceWindowChange(q *queuedRequest) {
+	if !f.started || !validWinchPayload(q.req.Payload) {
+		f.refuse(q)
 		return
 	}
 	if f.lastWinch != nil {
-		refuseRequest(f.lastWinch)
+		f.refuse(f.lastWinch)
 	}
-	f.lastWinch = req
+	f.lastWinch = q
 }
 
 // validWinchPayload reports whether p parses as a window-change payload whose columns and
@@ -223,6 +286,17 @@ func usablePtyDimension(v uint32) bool {
 	return v > 0 && v <= math.MaxUint16
 }
 
+// loopActsOn reports whether the session loop does anything with a request type other
+// than the ones the filter classifies itself. The loop answers every other type with a
+// failure, so its payload is not worth queueing.
+func loopActsOn(reqType string) bool {
+	switch reqType {
+	case "signal", "break", "auth-agent-req@openssh.com":
+		return true
+	}
+	return false
+}
+
 func refuseRequest(req *cryptossh.Request) {
 	req.Type = refusedRequestType
 	req.Payload = nil

client/ssh/server/session_limits_test.go

@@ -2,7 +2,9 @@ package server
 
 import (
 	"testing"
+	"time"
 
+	"github.com/gliderlabs/ssh"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	cryptossh "golang.org/x/crypto/ssh"
@@ -80,3 +82,105 @@ func TestValidWinchPayload_RejectsUnusableSizes(t *testing.T) {
 	}
 	assert.True(t, validWinchPayload(cryptossh.Marshal(struct{ Columns, Rows, Width, Height uint32 }{65535, 65535, 0, 0})), "largest uint16 size is usable")
 }
+
+// A client that stops the loop from reading must not make the server buffer more than
+// maxQueuedBytes of payload, however few requests that takes.
+func TestRequestFilter_QueueIsBoundedByBytes(t *testing.T) {
+	f := &requestFilter{}
+	payload := make([]byte, 32*1024)
+
+	accepted := 0
+	for f.enqueue(&cryptossh.Request{Type: "exec", Payload: payload}) {
+		accepted++
+		require.LessOrEqual(t, accepted, maxQueuedRequests, "queue must stop growing")
+	}
+	assert.LessOrEqual(t, f.queuedBytes, maxQueuedBytes, "queued payload bytes must stay under the limit")
+	assert.Equal(t, maxQueuedBytes/len(payload), accepted, "queue must hold payloads up to the byte limit")
+}
+
+// Request types the session loop only answers with a failure carry no useful payload, so
+// they are refused and cost nothing against the queue.
+func TestRequestFilter_UnhandledTypesDropPayload(t *testing.T) {
+	f := &requestFilter{}
+	req := &cryptossh.Request{Type: "x11-req", Payload: make([]byte, 32*1024)}
+	require.True(t, f.enqueue(req))
+
+	assert.Equal(t, refusedRequestType, req.Type, "unhandled type must be refused")
+	assert.Zero(t, f.queuedBytes, "refused request must not count its payload")
+}
+
+// Nothing drains window changes for a subsystem session, so a resize after one must be
+// refused for the rest of the session, even if a shell request follows.
+func TestRequestFilter_SubsystemSessionRefusesResize(t *testing.T) {
+	f := &requestFilter{}
+	winch := cryptossh.Marshal(struct{ Columns, Rows, Width, Height uint32 }{100, 40, 0, 0})
+
+	require.True(t, f.enqueue(&cryptossh.Request{Type: "subsystem", Payload: cryptossh.Marshal(struct{ Name string }{"sftp"})}))
+	require.True(t, f.enqueue(&cryptossh.Request{Type: "shell"}))
+	resize := &cryptossh.Request{Type: "window-change", Payload: winch}
+	require.True(t, f.enqueue(resize))
+	assert.Equal(t, refusedRequestType, resize.Type, "resize in a subsystem session must be refused")
+}
+
+// A loop blocked sending a resize that no handler reads is released by the drain, and the
+// drain ends when the loop closes the channel.
+func TestDrainWindowChanges_ReleasesBlockedSender(t *testing.T) {
+	winCh := make(chan ssh.Window, 1)
+	winCh <- ssh.Window{Width: 80, Height: 24}
+
+	sent := make(chan struct{})
+	go func() {
+		winCh <- ssh.Window{Width: 100, Height: 40}
+		close(sent)
+	}()
+	select {
+	case <-sent:
+		t.Fatal("sender should block while nothing reads the channel")
+	case <-time.After(50 * time.Millisecond):
+	}
+
+	drainWindowChanges(winCh)
+	select {
+	case <-sent:
+	case <-time.After(2 * time.Second):
+		t.Fatal("drain did not release the blocked sender")
+	}
+
+	// A drain that read one window change and exited would release the first sender but
+	// leave the loop blocked on the next resize.
+	for i := 0; i < 5; i++ {
+		select {
+		case winCh <- ssh.Window{Width: 120, Height: 48}:
+		case <-time.After(2 * time.Second):
+			t.Fatal("drain stopped consuming after the first window change")
+		}
+	}
+	close(winCh)
+}
+
+// A superseded resize keeps its place in the queue so its refusal is answered in order,
+// but must stop counting against the byte limit.
+func TestRequestFilter_CoalescedResizeReleasesQueuedBytes(t *testing.T) {
+	f := &requestFilter{}
+	require.True(t, f.enqueue(&cryptossh.Request{Type: "shell"}))
+	winch := cryptossh.Marshal(struct{ Columns, Rows, Width, Height uint32 }{100, 40, 0, 0})
+
+	for i := 0; i < 50; i++ {
+		require.True(t, f.enqueue(&cryptossh.Request{Type: "window-change", Payload: winch}))
+	}
+
+	assert.Equal(t, len(winch), f.queuedBytes, "only the newest resize should be charged")
+}
+
+// A large request the loop ignores is refused and its payload dropped, so it cannot close
+// the channel by pushing the queue over the byte limit.
+func TestRequestFilter_IgnoredLargeRequestDoesNotCloseChannel(t *testing.T) {
+	f := &requestFilter{}
+	for i := 0; i < maxQueuedBytes/len(make([]byte, 32*1024)); i++ {
+		require.True(t, f.enqueue(&cryptossh.Request{Type: "exec", Payload: make([]byte, 32*1024)}))
+	}
+
+	ignored := &cryptossh.Request{Type: "x11-req", Payload: make([]byte, 200*1024)}
+	assert.True(t, f.enqueue(ignored), "an ignored request must be refused rather than close the channel")
+	assert.Equal(t, refusedRequestType, ignored.Type)
+}