From 552202b8e32b8d23879f85948aa263fb809ec2e8 Mon Sep 17 00:00:00 2001 From: Dave Cheney Date: Wed, 5 Sep 2012 09:47:02 +1000 Subject: [PATCH] go.crypto/ssh: assorted close related fixes Fixes golang/go#3810. Fixes chanWriter Write after close behaviour bug. Fixes serverChan writePacket after close bug. Addresses final comments by agl on 6405064, plus various cleanups. R=agl, kardianos, gustav.paul, fullung CC=golang-dev https://golang.org/cl/6479056 --- ssh/channel.go | 86 ++++++++++++++++++++++++--------------------- ssh/client.go | 7 ++-- ssh/server.go | 2 +- ssh/session_test.go | 10 +----- ssh/transport.go | 6 ---- 5 files changed, 51 insertions(+), 60 deletions(-) diff --git a/ssh/channel.go b/ssh/channel.go index 97302db..f964a52 100644 --- a/ssh/channel.go +++ b/ssh/channel.go @@ -82,7 +82,7 @@ type channel struct { localId, remoteId uint32 remoteWin window maxPacket uint32 - isclosed uint32 // atomic bool, non zero if true + isClosed uint32 // atomic bool, non zero if true } func (c *channel) sendWindowAdj(n int) error { @@ -93,13 +93,20 @@ func (c *channel) sendWindowAdj(n int) error { return c.writePacket(marshal(msgChannelWindowAdjust, msg)) } -// sendEOF sends EOF to the server. RFC 4254 Section 5.3 +// sendEOF sends EOF to the remote side. RFC 4254 Section 5.3 func (c *channel) sendEOF() error { return c.writePacket(marshal(msgChannelEOF, channelEOFMsg{ PeersId: c.remoteId, })) } +// sendClose informs the remote side of our intent to close the channel. +func (c *channel) sendClose() error { + return c.conn.writePacket(marshal(msgChannelClose, channelCloseMsg{ + PeersId: c.remoteId, + })) +} + func (c *channel) sendChannelOpenFailure(reason RejectionReason, message string) error { reject := channelOpenFailureMsg{ PeersId: c.remoteId, @@ -121,11 +128,11 @@ func (c *channel) writePacket(b []byte) error { } func (c *channel) closed() bool { - return atomic.LoadUint32(&c.isclosed) > 0 + return atomic.LoadUint32(&c.isClosed) > 0 } func (c *channel) setClosed() bool { - return atomic.CompareAndSwapUint32(&c.isclosed, 0, 1) + return atomic.CompareAndSwapUint32(&c.isClosed, 0, 1) } type serverChan struct { @@ -136,10 +143,9 @@ type serverChan struct { serverConn *ServerConn myWindow uint32 - weClosed bool // incidates the close msg has been sent from our side theyClosed bool // indicates the close msg has been received from the remote side theySentEOF bool - dead bool + isDead uint32 err error pendingRequests []ChannelRequest @@ -247,7 +253,7 @@ func (edc extendedDataChannel) Write(data []byte) (n int, err error) { const headerLength = 13 // 1 byte message type, 4 bytes remoteId, 4 bytes extended message type, 4 bytes data length c := edc.c for len(data) > 0 { - space := uint32(min(int(c.maxPacket-headerLength), len(data))) + space := min(c.maxPacket-headerLength, len(data)) if space, err = c.getWindowSpace(space); err != nil { return 0, err } @@ -297,7 +303,7 @@ func (c *serverChan) read(data []byte) (n int, err error, windowAdjustment uint3 } for { - if c.theySentEOF || c.theyClosed || c.dead { + if c.theySentEOF || c.theyClosed || c.dead() { return 0, io.EOF, 0 } @@ -315,7 +321,7 @@ func (c *serverChan) read(data []byte) (n int, err error, windowAdjustment uint3 } if c.length > 0 { - tail := min(c.head+c.length, len(c.pendingData)) + tail := min(uint32(c.head+c.length), len(c.pendingData)) n = copy(data, c.pendingData[c.head:tail]) c.head += n c.length -= n @@ -341,24 +347,24 @@ func (c *serverChan) read(data []byte) (n int, err error, windowAdjustment uint3 // getWindowSpace takes, at most, max bytes of space from the peer's window. It // returns the number of bytes actually reserved. func (c *serverChan) getWindowSpace(max uint32) (uint32, error) { - var err error - // TODO(dfc) This lock and check of c.weClosed is necessary because unlike - // clientChan, c.weClosed is observed by more than one goroutine. - c.cond.L.Lock() - if c.dead || c.weClosed { - err = io.EOF - } - c.cond.L.Unlock() - if err != nil { - return 0, err + if c.dead() || c.closed() { + return 0, io.EOF } return c.remoteWin.reserve(max), nil } +func (c *serverChan) dead() bool { + return atomic.LoadUint32(&c.isDead) > 0 +} + +func (c *serverChan) setDead() { + atomic.StoreUint32(&c.isDead, 1) +} + func (c *serverChan) Write(data []byte) (n int, err error) { const headerLength = 9 // 1 byte message type, 4 bytes remoteId, 4 bytes data length for len(data) > 0 { - space := uint32(min(int(c.maxPacket-headerLength), len(data))) + space := min(c.maxPacket-headerLength, len(data)) if space, err = c.getWindowSpace(space); err != nil { return 0, err } @@ -384,6 +390,7 @@ func (c *serverChan) Write(data []byte) (n int, err error) { return } +// Close signals the intent to close the channel. func (c *serverChan) Close() error { c.serverConn.lock.Lock() defer c.serverConn.lock.Unlock() @@ -392,21 +399,12 @@ func (c *serverChan) Close() error { return c.serverConn.err } - if c.weClosed { + if !c.setClosed() { return errors.New("ssh: channel already closed") } - c.weClosed = true - return c.sendClose() } -// sendClose signals the intent to close the channel. -func (c *serverChan) sendClose() error { - return c.writePacket(marshal(msgChannelClose, channelCloseMsg{ - PeersId: c.remoteId, - })) -} - func (c *serverChan) AckRequest(ok bool) error { c.serverConn.lock.Lock() defer c.serverConn.lock.Unlock() @@ -491,32 +489,37 @@ func (c *clientChan) waitForChannelOpenResponse() error { return errors.New("ssh: unexpected packet") } +// Close signals the intent to close the channel. func (c *clientChan) Close() error { if !c.setClosed() { return errors.New("ssh: channel already closed") } c.stdout.eof() c.stderr.eof() - close(c.msg) - // TODO(dfc) step around channel.writePacket() because closed() is now true - return c.channel.conn.writePacket(marshal(msgChannelClose, channelCloseMsg{ - PeersId: c.remoteId, - })) + return c.sendClose() } // A chanWriter represents the stdin of a remote process. type chanWriter struct { *channel + // indicates the writer has been closed. eof is owned by the + // caller of Write/Close. + eof bool } // Write writes data to the remote process's standard input. func (w *chanWriter) Write(data []byte) (written int, err error) { const headerLength = 9 // 1 byte message type, 4 bytes remoteId, 4 bytes data length for len(data) > 0 { + if w.eof || w.closed() { + err = io.EOF + return + } // never send more data than maxPacket even if // there is sufficent window. - n := min(int(w.maxPacket-headerLength), len(data)) - n = int(w.remoteWin.reserve(uint32(n))) + n := min(w.maxPacket-headerLength, len(data)) + r := w.remoteWin.reserve(n) + n = r remoteId := w.remoteId packet := []byte{ msgChannelData, @@ -527,19 +530,20 @@ func (w *chanWriter) Write(data []byte) (written int, err error) { break } data = data[n:] - written += n + written += int(n) } return } -func min(a, b int) int { - if a < b { +func min(a uint32, b int) uint32 { + if a < uint32(b) { return a } - return b + return uint32(b) } func (w *chanWriter) Close() error { + w.eof = true return w.sendEOF() } diff --git a/ssh/client.go b/ssh/client.go index 2597907..dabc298 100644 --- a/ssh/client.go +++ b/ssh/client.go @@ -249,7 +249,8 @@ func (c *ClientConn) mainLoop() { ch.stderr.write(packet) } default: - switch msg := decode(packet).(type) { + msg := decode(packet) + switch msg := msg.(type) { case *channelOpenMsg: c.handleChanOpen(msg) case *channelOpenConfirmMsg: @@ -270,8 +271,7 @@ func (c *ClientConn) mainLoop() { return } ch.Close() - // TODO(dfc) may need to optimisically remove the - // channel before closing + close(ch.msg) c.chanList.remove(msg.PeersId) case *channelEOFMsg: ch, ok := c.getChan(msg.PeersId) @@ -502,5 +502,6 @@ func (c *chanList) closeAll() { continue } ch.Close() + close(ch.msg) } } diff --git a/ssh/server.go b/ssh/server.go index fc985a1..f359044 100644 --- a/ssh/server.go +++ b/ssh/server.go @@ -536,7 +536,7 @@ func (s *ServerConn) Accept() (Channel, error) { // TODO(dfc) s.lock protects s.channels but isn't being held here. for _, c := range s.channels { - c.dead = true + c.setDead() c.handleData(nil) } diff --git a/ssh/session_test.go b/ssh/session_test.go index 7fc52ed..f70ae83 100644 --- a/ssh/session_test.go +++ b/ssh/session_test.go @@ -47,7 +47,7 @@ func dial(handler serverType, t *testing.T) *ClientConn { done := make(chan struct{}) for { ch, err := conn.Accept() - if err == io.EOF { + if err == io.EOF || err == io.ErrUnexpectedEOF { return } // We sometimes get ECONNRESET rather than EOF. @@ -389,10 +389,6 @@ func TestServerStdoutRespectsMaxPacketSize(t *testing.T) { } func TestClientCannotSendAfterEOF(t *testing.T) { - // TODO(dfc) currently writes succeed after Close() - t.Logf("test skipped") - return - conn := dial(exitWithoutSignalOrStatus, t) defer conn.Close() session, err := conn.NewSession() @@ -628,10 +624,6 @@ func discardHandler(ch *serverChan, t *testing.T) { // grow the window to avoid being fooled by // the initial 1 << 14 window. ch.sendWindowAdj(1024 * 1024) - // TODO(dfc) io.Copy can return a non EOF error here - // because closed channel errors can leak here if the - // read from ch causes a window adjustment after the - // remote has signaled close. io.Copy(ioutil.Discard, ch) } diff --git a/ssh/transport.go b/ssh/transport.go index 73b9783..424b6c9 100644 --- a/ssh/transport.go +++ b/ssh/transport.go @@ -194,12 +194,6 @@ func (w *writer) writePacket(packet []byte) error { return w.Flush() } -// Send a message to the remote peer -func (t *transport) sendMessage(typ uint8, msg interface{}) error { - packet := marshal(typ, msg) - return t.writePacket(packet) -} - func newTransport(conn net.Conn, rand io.Reader) *transport { return &transport{ reader: reader{