From 08b924f7398a8e31ae04a78a744d87fc0376f37f Mon Sep 17 00:00:00 2001 From: Viktor Liu <17948409+lixmal@users.noreply.github.com> Date: Mon, 26 Oct 2026 23:04:32 +0900 Subject: [PATCH] [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 | 9 +- client/ssh/server/session_hardening_test.go | 62 ++++++++++ client/ssh/server/session_limits.go | 130 +++++++++++++++----- client/ssh/server/session_limits_test.go | 104 ++++++++++++++++ 4 files changed, 276 insertions(+), 29 deletions(-) diff --git a/client/ssh/server/session_handlers.go b/client/ssh/server/session_handlers.go index 0e531bb96..45b0fc3be 100644 --- a/client/ssh/server/session_handlers.go +++ b/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 { diff --git a/client/ssh/server/session_hardening_test.go b/client/ssh/server/session_hardening_test.go index ed107a32a..f60bb3c3d 100644 --- a/client/ssh/server/session_hardening_test.go +++ b/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") +} diff --git a/client/ssh/server/session_limits.go b/client/ssh/server/session_limits.go index 68850e4bd..4e004b00b 100644 --- a/client/ssh/server/session_limits.go +++ b/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 diff --git a/client/ssh/server/session_limits_test.go b/client/ssh/server/session_limits_test.go index 8c2a35d37..93f4d5f48 100644 --- a/client/ssh/server/session_limits_test.go +++ b/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) +}