ssh: fix deadlock on unexpected channel responses

Previously, channel.handlePacket sent channelRequestSuccess and
channelRequestFailure messages to ch.msg unconditionally via the default
arm of its type switch. Because ch.msg is a bounded buffer (chanSize),
a peer that sends a burst of unsolicited channel request responses for
an open, idle channel fills the buffer and blocks the mux read loop on
the next send. That stalls all packet processing on the connection,
and because readLoop then backs up on t.incoming, closing the
underlying net.Conn does not unblock either goroutine: user code
observes Close() returning promptly while Wait() hangs and the mux,
readLoop, and kexLoop goroutines leak permanently.

This change mirrors the fix for the mux-level SendRequest path: a
sentRequestPending atomic gate is set while a SendRequest with
WantReply is in flight, handlePacket drops responses when the gate is
closed, and uses a non-blocking send otherwise. SendRequest drains
any spurious response that slipped through before discarding it, so
the caller always observes the reply to its own request.

This aligns with OpenSSH, which silently ignores channel confirm
messages that do not match a pending request.

Fixes golang/go#79564
Fixes CVE-2026-39830

Change-Id: I15e2add4bf7876bb0c6f921f8b57203d97e83f47
Reviewed-on: https://go-review.googlesource.com/c/crypto/+/781664
LUCI-TryBot-Result: golang-scoped@luci-project-accounts.iam.gserviceaccount.com <golang-scoped@luci-project-accounts.iam.gserviceaccount.com>
Auto-Submit: Neal Patel <nealpatel@google.com>
Reviewed-by: Neal Patel <nealpatel@google.com>
Reviewed-by: Roland Shoemaker <roland@golang.org>

Author: drakkan

ssh/channel.go

@@ -11,6 +11,7 @@ import (
 	"io"
 	"log"
 	"sync"
+	"sync/atomic"
 )
 
 const (
@@ -183,6 +184,12 @@ type channel struct {
 	// with WantReply=true outstanding.  This lock is held by a
 	// goroutine that has such an outgoing request pending.
 	sentRequestMu sync.Mutex
+	// sentRequestPending is set to true while a SendRequest call with
+	// WantReply=true is in flight. handlePacket uses it as a gate: responses
+	// arriving while no request is pending are dropped to prevent a
+	// misbehaving peer from stalling the mux read loop by filling ch.msg
+	// with unsolicited channelRequestSuccess/Failure messages.
+	sentRequestPending atomic.Bool
 
 	incomingRequests chan *Request
 
@@ -466,6 +473,18 @@ func (ch *channel) handlePacket(packet []byte) error {
 		}
 
 		ch.incomingRequests <- &req
+	case *channelRequestSuccessMsg, *channelRequestFailureMsg:
+		// Drop responses that arrive when no SendRequest is waiting, to
+		// prevent a malicious peer from filling ch.msg and stalling the
+		// mux read loop. The non-blocking send additionally protects the
+		// loop if a well-behaved caller is slow to read.
+		if !ch.sentRequestPending.Load() {
+			return nil
+		}
+		select {
+		case ch.msg <- msg:
+		default:
+		}
 	default:
 		ch.msg <- msg
 	}
@@ -602,6 +621,24 @@ func (ch *channel) SendRequest(name string, wantReply bool, payload []byte) (boo
 	if wantReply {
 		ch.sentRequestMu.Lock()
 		defer ch.sentRequestMu.Unlock()
+
+		// Open the gate so that responses arriving while this request is in
+		// flight are allowed to reach ch.msg. Responses arriving while no
+		// request is pending are dropped by handlePacket.
+		ch.sentRequestPending.Store(true)
+		defer ch.sentRequestPending.Store(false)
+
+		// Drain any spurious responses that may have been buffered. This
+		// prevents a previously buffered unexpected response from being
+		// consumed instead of the actual response for this request.
+	drain:
+		for {
+			select {
+			case <-ch.msg:
+			default:
+				break drain
+			}
+		}
 	}
 
 	msg := channelRequestMsg{

ssh/mux_test.go

@@ -1135,3 +1135,234 @@ func TestMuxGlobalResponseAcceptedWhilePending(t *testing.T) {
 	clientMux.Close()
 	<-serverDone
 }
+
+func TestChannelUnexpectedResponsesDiscarded(t *testing.T) {
+	// A malicious peer that spams channelRequestSuccess/Failure messages
+	// for an open, idle channel must not be able to stall the mux read
+	// loop by filling ch.msg. After the flood, the channel must still be
+	// usable: a subsequent legitimate SendRequest receives its reply.
+	clientMux, serverMux := muxPair()
+	defer serverMux.Close()
+	defer clientMux.Close()
+
+	serverRes := make(chan *channel, 1)
+	go func() {
+		newCh, ok := <-serverMux.incomingChannels
+		if !ok {
+			close(serverRes)
+			return
+		}
+		c, _, err := newCh.Accept()
+		if err != nil {
+			close(serverRes)
+			return
+		}
+		serverRes <- c.(*channel)
+	}()
+
+	clientCh, err := clientMux.openChannel("chan", nil)
+	if err != nil {
+		t.Fatalf("openChannel: %v", err)
+	}
+	serverCh := <-serverRes
+	if serverCh == nil {
+		t.Fatal("server did not accept channel")
+	}
+
+	// Spam many unsolicited success/failure responses. More than chanSize
+	// to ensure ch.msg would overflow without the pending-gate.
+	const spam = chanSize * 4
+	done := make(chan error, 1)
+	go func() {
+		for i := range spam {
+			if err := serverCh.ackRequest(i%2 == 0); err != nil {
+				done <- fmt.Errorf("ackRequest %d: %w", i, err)
+				return
+			}
+		}
+		// Echo any legitimate request back.
+		for req := range serverCh.incomingRequests {
+			if req.WantReply {
+				if err := req.Reply(true, append([]byte("reply:"), req.Payload...)); err != nil {
+					done <- fmt.Errorf("reply: %w", err)
+					return
+				}
+			}
+		}
+		done <- nil
+	}()
+
+	// If the flood had wedged the mux loop, this SendRequest would never
+	// receive a reply.
+	ok, err := clientCh.SendRequest("ping", true, []byte("hello"))
+	if err != nil {
+		t.Fatalf("SendRequest: %v", err)
+	}
+	if !ok {
+		t.Fatal("expected success reply")
+	}
+
+	// Clean up so the server goroutine can exit.
+	clientCh.Close()
+	serverCh.Close()
+	if err := <-done; err != nil {
+		if !errors.Is(err, io.EOF) {
+			t.Fatal(err)
+		}
+	}
+}
+
+func TestChannelConcurrentRequests(t *testing.T) {
+	writer, reader, mux := channelPair(t)
+	defer writer.Close()
+	defer reader.Close()
+	defer mux.Close()
+
+	serverDone := make(chan struct{})
+	go func() {
+		defer close(serverDone)
+		for req := range writer.incomingRequests {
+			if req.WantReply {
+				req.Reply(true, append([]byte("reply:"), req.Payload...))
+			}
+		}
+	}()
+
+	const numRequests = 50
+	var wg sync.WaitGroup
+	wg.Add(numRequests)
+	errCh := make(chan error, numRequests)
+
+	for i := 0; i < numRequests; i++ {
+		go func(id int) {
+			defer wg.Done()
+			payload := []byte(fmt.Sprintf("req-%d", id))
+			ok, err := reader.SendRequest("echo", true, payload)
+			if err != nil {
+				errCh <- fmt.Errorf("req %d: %v", id, err)
+				return
+			}
+			if !ok {
+				errCh <- fmt.Errorf("req %d: expected success", id)
+			}
+		}(i)
+	}
+
+	wg.Wait()
+	close(errCh)
+
+	for err := range errCh {
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	reader.Close()
+	writer.Close()
+	<-serverDone
+}
+
+func TestChannelResponseDroppedWhenIdle(t *testing.T) {
+	// A spurious response arriving while no SendRequest is pending must
+	// be dropped rather than buffered in ch.msg.
+	writer, reader, mux := channelPair(t)
+	defer writer.Close()
+	defer reader.Close()
+	defer mux.Close()
+
+	// Server sends an unsolicited reply, then a request so we can
+	// synchronise: once the client observes the request, the mux loop has
+	// necessarily processed (and dropped) the prior spurious reply.
+	errCh := make(chan error, 1)
+	go func() {
+		if err := writer.ackRequest(true); err != nil {
+			errCh <- err
+			return
+		}
+		if _, err := writer.SendRequest("sync", false, nil); err != nil {
+			errCh <- err
+			return
+		}
+		errCh <- nil
+	}()
+
+	req := <-reader.incomingRequests
+	if req.Type != "sync" {
+		t.Fatalf("unexpected request type %q", req.Type)
+	}
+
+	if n := len(reader.msg); n != 0 {
+		t.Fatalf("ch.msg should be empty after idle drop, has %d entries", n)
+	}
+
+	if err := <-errCh; err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestChannelStaleResponseDrained(t *testing.T) {
+	// Simulate a stale response sitting in ch.msg (e.g. a response that
+	// slipped through the pending-gate on a prior SendRequest that exited
+	// without consuming it). The drain step in the next SendRequest must
+	// discard it so the caller receives the correct reply.
+	writer, reader, mux := channelPair(t)
+	defer writer.Close()
+	defer reader.Close()
+	defer mux.Close()
+
+	reader.msg <- &channelRequestSuccessMsg{PeersID: reader.remoteId}
+
+	serverDone := make(chan struct{})
+	go func() {
+		defer close(serverDone)
+		for req := range writer.incomingRequests {
+			if req.WantReply {
+				req.Reply(false, append([]byte("nack:"), req.Payload...))
+			}
+		}
+	}()
+
+	ok, err := reader.SendRequest("test", true, []byte("hello"))
+	if err != nil {
+		t.Fatalf("SendRequest: %v", err)
+	}
+	// If the stale success had been consumed, ok would be true.
+	if ok {
+		t.Fatal("got stale success response; drain did not remove it")
+	}
+
+	reader.Close()
+	writer.Close()
+	<-serverDone
+}
+
+func TestChannelResponseAcceptedWhilePending(t *testing.T) {
+	// Positive control: when a SendRequest is actually pending, the
+	// response must be delivered (the gate is open).
+	writer, reader, mux := channelPair(t)
+	defer writer.Close()
+	defer reader.Close()
+	defer mux.Close()
+
+	serverDone := make(chan struct{})
+	go func() {
+		defer close(serverDone)
+		for req := range writer.incomingRequests {
+			if req.WantReply {
+				req.Reply(true, nil)
+			}
+		}
+	}()
+
+	ok, err := reader.SendRequest("ping", true, nil)
+	if err != nil {
+		t.Fatalf("SendRequest: %v", err)
+	}
+	if !ok {
+		t.Fatal("expected success")
+	}
+
+	reader.Close()
+	writer.Close()
+	<-serverDone
+}