From f2b1b98d67e15bae02d066a04c39dd462a333ebf Mon Sep 17 00:00:00 2001 From: Nicola Murino Date: Sun, 25 Jan 2026 19:08:01 +0100 Subject: [PATCH 1/2] ssh: fix deadlock on unexpected global responses Previously, the mux implementation handled global request responses by blocking until the response could be sent to the globalResponses channel. Since this channel has a buffer size of 1, unsolicited responses from a server (or responses arriving after a timeout) would fill the buffer. Subsequent unsolicited responses would block handleGlobalPacket, stalling the entire connection's read loop and causing a denial of service. This change modifies handleGlobalPacket to use a non-blocking send. If no goroutine is waiting for a response (or the buffer is full), the message is dropped. This aligns with OpenSSH behavior, which ignores unexpected global responses. Additionally, SendRequest now drains the globalResponses channel after acquiring the mutex but before sending the request. This ensures that any stale responses or "spam" buffered just before the lock was acquired are discarded, preventing race conditions where a legitimate request might otherwise consume an unrelated response. This issue was found during a security audit by NCC Group Cryptography Services, sponsored by Teleport. Fixes golang/go#79564 Fixes CVE-2026-39830 Change-Id: Ia0c46355203d557eadcd432c10b87c8a044e1089 Reviewed-on: https://go-review.googlesource.com/c/crypto/+/781640 Reviewed-by: Roland Shoemaker Reviewed-by: Neal Patel LUCI-TryBot-Result: golang-scoped@luci-project-accounts.iam.gserviceaccount.com --- ssh/mux.go | 36 ++++++- ssh/mux_test.go | 254 ++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 286 insertions(+), 4 deletions(-) diff --git a/ssh/mux.go b/ssh/mux.go index d2d24c635d..3bc4afbd0f 100644 --- a/ssh/mux.go +++ b/ssh/mux.go @@ -91,9 +91,10 @@ type mux struct { incomingChannels chan NewChannel - globalSentMu sync.Mutex - globalResponses chan interface{} - incomingRequests chan *Request + globalSentMu sync.Mutex + globalSentPending atomic.Bool + globalResponses chan interface{} + incomingRequests chan *Request errCond *sync.Cond err error @@ -141,6 +142,24 @@ func (m *mux) SendRequest(name string, wantReply bool, payload []byte) (bool, [] if wantReply { m.globalSentMu.Lock() defer m.globalSentMu.Unlock() + + // Open the gate so that responses arriving while this request is in + // flight are allowed to reach globalResponses. Any response arriving + // while no request is pending is dropped by handleGlobalPacket. + m.globalSentPending.Store(true) + defer m.globalSentPending.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 <-m.globalResponses: + default: + break drain + } + } } if err := m.sendMessage(globalRequestMsg{ @@ -267,7 +286,16 @@ func (m *mux) handleGlobalPacket(packet []byte) error { mux: m, } case *globalRequestSuccessMsg, *globalRequestFailureMsg: - m.globalResponses <- msg + // Drop responses that arrive when no SendRequest is waiting, to + // prevent a malicious peer from staging responses for a future + // caller. + if !m.globalSentPending.Load() { + return nil + } + select { + case m.globalResponses <- msg: + default: + } default: panic(fmt.Sprintf("not a global message %#v", msg)) } diff --git a/ssh/mux_test.go b/ssh/mux_test.go index 21f0ac3e32..18e23991bf 100644 --- a/ssh/mux_test.go +++ b/ssh/mux_test.go @@ -837,3 +837,257 @@ func TestDebug(t *testing.T) { t.Error("transport debug switched on") } } + +func TestMuxUnexpectedGlobalResponsesDiscarded(t *testing.T) { + clientPipe, serverPipe := memPipe() + client := newMux(clientPipe) + defer serverPipe.Close() + defer client.Close() + + done := make(chan error, 1) + go func() { + // Send multiple unexpected global responses, this should not block the + // globalResponses channel. + for i := range 5 { + err := serverPipe.writePacket(Marshal(globalRequestSuccessMsg{ + Data: []byte{byte(i)}, + })) + if err != nil { + done <- fmt.Errorf("send success msg %d: %w", i, err) + return + } + } + for i := range 5 { + err := serverPipe.writePacket(Marshal(globalRequestFailureMsg{ + Data: []byte{byte(i)}, + })) + if err != nil { + done <- fmt.Errorf("send failure msg %d: %w", i, err) + return + } + } + + // Now send a global request and wait for the response. This + // verifies the mux is still processing packets. + err := serverPipe.writePacket(Marshal(globalRequestMsg{ + Type: "keepalive@golang.org", + WantReply: true, + Data: nil, + })) + if err != nil { + done <- fmt.Errorf("send global request: %w", err) + return + } + + packet, err := serverPipe.readPacket() + if err != nil { + done <- fmt.Errorf("read packet: %w", err) + return + } + decoded, err := decode(packet) + if err != nil { + done <- fmt.Errorf("decode: %w", err) + return + } + switch decoded.(type) { + case *globalRequestSuccessMsg, *globalRequestFailureMsg: + // Expected response + default: + done <- fmt.Errorf("unexpected packet type: %T", decoded) + return + } + done <- nil + }() + + // Handle the incoming request from the server and reply + req, ok := <-client.incomingRequests + if !ok { + t.Fatal("incomingRequests channel closed unexpectedly") + } + if req.Type != "keepalive@golang.org" { + t.Fatalf("unexpected request type: %s", req.Type) + } + if err := req.Reply(true, nil); err != nil { + t.Fatalf("Reply: %v", err) + } + + if err := <-done; err != nil { + t.Fatal(err) + } +} + +func TestMuxConcurrentGlobalRequests(t *testing.T) { + clientMux, serverMux := muxPair() + defer serverMux.Close() + defer clientMux.Close() + + const numRequests = 50 + + serverDone := make(chan struct{}) + go func() { + defer close(serverDone) + for r := range serverMux.incomingRequests { + if r.WantReply { + replyData := append([]byte("reply:"), r.Payload...) + r.Reply(true, replyData) + } + } + }() + + var clientWg sync.WaitGroup + clientWg.Add(numRequests) + + errCh := make(chan error, numRequests) + + for i := range numRequests { + go func(id int) { + defer clientWg.Done() + + payloadStr := fmt.Sprintf("req-%d", id) + payload := []byte(payloadStr) + + // This call blocks until the globalSentMu is acquired. + // The mutex ensures that even with many concurrent attempts, + // the "drain" and "send" logic happens atomically per request. + ok, data, err := clientMux.SendRequest("echo", true, payload) + if err != nil { + errCh <- fmt.Errorf("req %d error: %v", id, err) + return + } + if !ok { + errCh <- fmt.Errorf("req %d failed (want success)", id) + return + } + + expected := "reply:" + payloadStr + if string(data) != expected { + errCh <- fmt.Errorf("req %d mismatch: got %q, want %q", id, string(data), expected) + } + }(i) + } + + clientWg.Wait() + close(errCh) + + for err := range errCh { + if err != nil { + t.Fatal(err) + } + } + + clientMux.Close() + <-serverDone +} + +func TestMuxGlobalResponseDroppedWhenIdle(t *testing.T) { + clientPipe, serverPipe := memPipe() + clientMux := newMux(clientPipe) + defer serverPipe.Close() + defer clientMux.Close() + + errCh := make(chan error, 1) + go func() { + // Send a spurious response while no SendRequest is pending. + if err := serverPipe.writePacket(Marshal(globalRequestSuccessMsg{ + Data: []byte("spurious"), + })); err != nil { + errCh <- fmt.Errorf("send spurious: %w", err) + return + } + // Follow with a global request; once the client observes this on + // incomingRequests, the mux loop has necessarily processed (and + // dropped) the prior spurious response. + if err := serverPipe.writePacket(Marshal(globalRequestMsg{ + Type: "sync@example.com", + WantReply: false, + })); err != nil { + errCh <- fmt.Errorf("send sync request: %w", err) + return + } + errCh <- nil + }() + + if err := <-errCh; err != nil { + t.Fatal(err) + } + + req, ok := <-clientMux.incomingRequests + if !ok { + t.Fatal("incomingRequests closed unexpectedly") + } + if req.Type != "sync@example.com" { + t.Fatalf("unexpected sync request type %q", req.Type) + } + + // The spurious response preceded the sync request, so by now the mux + // loop has processed it. The pending-gate must have caused it to be + // dropped rather than buffered. + if n := len(clientMux.globalResponses); n != 0 { + t.Fatalf("globalResponses buffer should be empty after idle drop, has %d entries", n) + } +} + +func TestMuxStaleResponseDrained(t *testing.T) { + // Simulate a stale response sitting in globalResponses (e.g. a response + // that slipped in 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. + clientMux, serverMux := muxPair() + defer serverMux.Close() + defer clientMux.Close() + + clientMux.globalResponses <- &globalRequestSuccessMsg{Data: []byte("stale")} + + serverDone := make(chan struct{}) + go func() { + defer close(serverDone) + for req := range serverMux.incomingRequests { + if req.WantReply { + req.Reply(true, append([]byte("reply:"), req.Payload...)) + } + } + }() + + ok, data, err := clientMux.SendRequest("test", true, []byte("hello")) + if err != nil { + t.Fatalf("request failed: %v", err) + } + if !ok { + t.Fatal("expected success response") + } + if string(data) != "reply:hello" { + t.Fatalf("got %q, want %q (drain did not remove stale response)", data, "reply:hello") + } + + clientMux.Close() + <-serverDone +} + +func TestMuxGlobalResponseAcceptedWhilePending(t *testing.T) { + // Positive control: when a SendRequest is actually pending, the + // response must be delivered (the gate is open). + clientMux, serverMux := muxPair() + defer serverMux.Close() + defer clientMux.Close() + + serverDone := make(chan struct{}) + go func() { + defer close(serverDone) + for req := range serverMux.incomingRequests { + if req.WantReply { + req.Reply(true, []byte("pong")) + } + } + }() + + ok, data, err := clientMux.SendRequest("ping", true, nil) + if err != nil { + t.Fatalf("SendRequest: %v", err) + } + if !ok || string(data) != "pong" { + t.Fatalf("unexpected response: ok=%v data=%q", ok, data) + } + + clientMux.Close() + <-serverDone +} From 779cc46a4470531f995efe35c54f817f039d5128 Mon Sep 17 00:00:00 2001 From: Nicola Date: Tue, 21 Apr 2026 21:43:00 +0200 Subject: [PATCH 2/2] 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 Auto-Submit: Neal Patel Reviewed-by: Neal Patel Reviewed-by: Roland Shoemaker --- ssh/channel.go | 37 ++++++++ ssh/mux_test.go | 231 ++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 268 insertions(+) diff --git a/ssh/channel.go b/ssh/channel.go index cc0bb7ab64..49c301567f 100644 --- a/ssh/channel.go +++ b/ssh/channel.go @@ -11,6 +11,7 @@ import ( "io" "log" "sync" + "sync/atomic" ) const ( @@ -177,6 +178,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 @@ -460,6 +467,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 } @@ -586,6 +605,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{ diff --git a/ssh/mux_test.go b/ssh/mux_test.go index 18e23991bf..0654cbec42 100644 --- a/ssh/mux_test.go +++ b/ssh/mux_test.go @@ -1091,3 +1091,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 +}