ssh: drain stderr on forwarded TCP and Unix channels

tcpListener.Accept, unixListener.Accept, Client.dial and
Client.dialStreamLocal only read the main stream of the channels they
return. Data sent by the peer on the extended (stderr) stream
accumulates in the extPending buffer and its window credit is only
returned by ReadExtended, which is never called. Since the window is
shared between the two streams, a misbehaving peer can pin up to 2 MiB
per channel and eventually stall it; well-behaved peers never send
stderr on these channel types.

Drain the stderr stream into io.Discard, mirroring CL 783720 which
fixed the same issue in the agent forwarders.

The drain takes precedence over reading Stderr through a type assertion
on the returned net.Conn, which was never a supported use. For
comparison, OpenSSH refuses extended data on these channel types and
drops the payload without buffering it.

Updates golang/go#80333

Change-Id: I4f1445eee9ce1b56ca62cec342812d63cfdd3e13
Reviewed-on: https://go-review.googlesource.com/c/crypto/+/802900
Reviewed-by: Filippo Valsorda <filippo@golang.org>
Reviewed-by: Carlos Amedee <carlos@golang.org>
Reviewed-by: Cherry Mui <cherryyz@google.com>
LUCI-TryBot-Result: golang-scoped@luci-project-accounts.iam.gserviceaccount.com <golang-scoped@luci-project-accounts.iam.gserviceaccount.com>

ssh/streamlocal.go

diff --git a/ssh/streamlocal.go b/ssh/streamlocal.go
index 152470f..8e997da 100644
--- a/ssh/streamlocal.go
+++ b/ssh/streamlocal.go
@@ -58,6 +58,7 @@
 		return nil, err
 	}
 	go DiscardRequests(in)
+	go io.Copy(io.Discard, ch.Stderr())
 	return ch, err
 }
 
@@ -79,6 +80,7 @@
 		return nil, err
 	}
 	go DiscardRequests(incoming)
+	go io.Copy(io.Discard, ch.Stderr())
 
 	return &chanConn{
 		Channel: ch,

ssh/tcpip.go

diff --git a/ssh/tcpip.go b/ssh/tcpip.go
index 78c41fe..213d8a6 100644
--- a/ssh/tcpip.go
+++ b/ssh/tcpip.go
@@ -332,6 +332,7 @@
 		return nil, err
 	}
 	go DiscardRequests(incoming)
+	go io.Copy(io.Discard, ch.Stderr())
 
 	return &chanConn{
 		Channel: ch,
@@ -495,6 +496,7 @@
 		return nil, err
 	}
 	go DiscardRequests(in)
+	go io.Copy(io.Discard, ch.Stderr())
 	return ch, nil
 }
 

ssh/tcpip_test.go

diff --git a/ssh/tcpip_test.go b/ssh/tcpip_test.go
index 4d85114..5ec7ed9 100644
--- a/ssh/tcpip_test.go
+++ b/ssh/tcpip_test.go
@@ -6,6 +6,7 @@
 
 import (
 	"context"
+	"io"
 	"net"
 	"testing"
 	"time"
@@ -51,3 +52,213 @@
 		t.Errorf("DialContext: got nil error, expected %v", context.DeadlineExceeded)
 	}
 }
+
+// forwardingPair establishes a client/server connection pair over an
+// in-memory pipe. The server grants all global requests and accepts all
+// channels opened by the client, delivering them on the returned channel.
+func forwardingPair(t *testing.T) (*Client, *ServerConn, <-chan Channel) {
+	t.Helper()
+
+	c1, c2, err := netPipe()
+	if err != nil {
+		t.Fatalf("netPipe: %v", err)
+	}
+	serverDone := make(chan struct{})
+	t.Cleanup(func() { <-serverDone })
+	t.Cleanup(func() {
+		c1.Close()
+		c2.Close()
+	})
+
+	serverConf := ServerConfig{
+		NoClientAuth: true,
+	}
+	serverConf.AddHostKey(testSigners["rsa"])
+	incoming := make(chan *ServerConn, 1)
+	serverChans := make(chan Channel, 1)
+	go func() {
+		defer close(serverDone)
+		conn, chans, reqs, err := NewServerConn(c1, &serverConf)
+		incoming <- conn
+		if err != nil {
+			t.Errorf("NewServerConn error: %v", err)
+			return
+		}
+		go func() {
+			for r := range reqs {
+				r.Reply(true, nil)
+			}
+		}()
+		for newCh := range chans {
+			ch, chanReqs, err := newCh.Accept()
+			if err != nil {
+				t.Errorf("Accept: %v", err)
+				continue
+			}
+			go DiscardRequests(chanReqs)
+			serverChans <- ch
+		}
+	}()
+
+	conf := ClientConfig{
+		HostKeyCallback: InsecureIgnoreHostKey(),
+	}
+	conn, chans, reqs, err := NewClientConn(c2, "", &conf)
+	if err != nil {
+		t.Fatalf("NewClientConn: %v", err)
+	}
+	client := NewClient(conn, chans, reqs)
+	t.Cleanup(func() { client.Close() })
+
+	server := <-incoming
+	if server == nil {
+		t.Fatal("Unable to get server")
+	}
+	t.Cleanup(func() { server.Close() })
+
+	return client, server, serverChans
+}
+
+// floodStderr writes more than the default channel receive window (2 MiB)
+// to w, which must be the stderr stream of an SSH channel. Stderr data is
+// buffered and its window credit is only returned when the stream is
+// read, so if the peer does not drain the stream the write blocks forever
+// once the remote window is exhausted.
+func floodStderr(t *testing.T, w io.Writer) {
+	t.Helper()
+
+	const payload = 4 * 1024 * 1024
+	done := make(chan error, 1)
+	go func() {
+		buf := make([]byte, 32*1024)
+		remaining := payload
+		for remaining > 0 {
+			n := min(len(buf), remaining)
+			nw, err := w.Write(buf[:n])
+			if err != nil {
+				done <- err
+				return
+			}
+			remaining -= nw
+		}
+		done <- nil
+	}()
+
+	select {
+	case err := <-done:
+		if err != nil {
+			t.Fatalf("write to stderr: %v", err)
+		}
+	case <-time.After(10 * time.Second):
+		t.Fatal("write to stderr blocked: the stderr stream is not drained")
+	}
+}
+
+// verifyMainStream checks that the main stream is still usable after
+// stderr traffic by sending a payload from the server channel and
+// reading it back from the client connection.
+func verifyMainStream(t *testing.T, serverCh Channel, clientConn net.Conn) {
+	t.Helper()
+
+	want := []byte("main stream after stderr flood")
+	if _, err := serverCh.Write(want); err != nil {
+		t.Fatalf("write to main stream: %v", err)
+	}
+	got := make([]byte, len(want))
+	if _, err := io.ReadFull(clientConn, got); err != nil {
+		t.Fatalf("read from main stream: %v", err)
+	}
+	if string(got) != string(want) {
+		t.Fatalf("read %q from main stream, want %q", got, want)
+	}
+}
+
+func TestClientDialDiscardsStderr(t *testing.T) {
+	for _, network := range []string{"tcp", "unix"} {
+		t.Run(network, func(t *testing.T) {
+			client, _, serverChans := forwardingPair(t)
+
+			addr := "127.0.0.1:8022"
+			if network == "unix" {
+				addr = "/tmp/forwarded.sock"
+			}
+			conn, err := client.Dial(network, addr)
+			if err != nil {
+				t.Fatalf("Dial(%q, %q): %v", network, addr, err)
+			}
+			defer conn.Close()
+			serverCh := <-serverChans
+			defer serverCh.Close()
+
+			floodStderr(t, serverCh.Stderr())
+			verifyMainStream(t, serverCh, conn)
+		})
+	}
+}
+
+func TestClientListenerDiscardsStderr(t *testing.T) {
+	for _, network := range []string{"tcp", "unix"} {
+		t.Run(network, func(t *testing.T) {
+			client, server, _ := forwardingPair(t)
+
+			var (
+				ln      net.Listener
+				payload []byte
+				chType  string
+				err     error
+			)
+			if network == "tcp" {
+				ln, err = client.Listen("tcp", "127.0.0.1:8022")
+				if err != nil {
+					t.Fatalf("Listen: %v", err)
+				}
+				chType = "forwarded-tcpip"
+				payload = Marshal(&forwardedTCPPayload{
+					Addr:       "127.0.0.1",
+					Port:       8022,
+					OriginAddr: "127.0.0.5",
+					OriginPort: 1234,
+				})
+			} else {
+				ln, err = client.ListenUnix("/tmp/forwarded.sock")
+				if err != nil {
+					t.Fatalf("ListenUnix: %v", err)
+				}
+				chType = "forwarded-streamlocal@openssh.com"
+				payload = Marshal(&forwardedStreamLocalPayload{
+					SocketPath: "/tmp/forwarded.sock",
+				})
+			}
+			defer ln.Close()
+
+			// OpenChannel does not return until the channel is
+			// accepted, so run Accept concurrently.
+			accepted := make(chan net.Conn, 1)
+			go func() {
+				conn, err := ln.Accept()
+				if err != nil {
+					t.Errorf("Accept: %v", err)
+					close(accepted)
+					return
+				}
+				accepted <- conn
+			}()
+
+			serverCh, reqs, err := server.OpenChannel(chType, payload)
+			if err != nil {
+				t.Fatalf("OpenChannel(%q): %v", chType, err)
+			}
+			go DiscardRequests(reqs)
+			defer serverCh.Close()
+
+			conn := <-accepted
+			if conn == nil {
+				t.Fatal("no connection accepted")
+			}
+			defer conn.Close()
+
+			floodStderr(t, serverCh.Stderr())
+			verifyMainStream(t, serverCh, conn)
+		})
+	}
+}