diff --git a/ssh/channel.go b/ssh/channel.go index a65c79d..4567f6a 100644 --- a/ssh/channel.go +++ b/ssh/channel.go @@ -499,19 +499,20 @@ func (ch *channel) handlePacket(packet []byte) error { func (m *mux) newChannel(chanType string, direction channelDirection, extraData []byte) *channel { ch := &channel{ - remoteWin: window{Cond: newCond()}, - myWindow: channelWindowSize, - pending: newBuffer(), - extPending: newBuffer(), - direction: direction, - incomingRequests: make(chan *Request, chanSize), - msg: make(chan interface{}, chanSize), - chanType: chanType, - extraData: extraData, - mux: m, - packetPool: make(map[uint32][]byte), + remoteWin: window{Cond: newCond()}, + myWindow: channelWindowSize, + maxIncomingPayload: channelMaxPacket, + pending: newBuffer(), + extPending: newBuffer(), + direction: direction, + incomingRequests: make(chan *Request, chanSize), + msg: make(chan interface{}, chanSize), + chanType: chanType, + extraData: extraData, + mux: m, + packetPool: make(map[uint32][]byte), } - ch.localId = m.chanList.add(ch) + m.chanList.add(ch) return ch } @@ -535,7 +536,6 @@ func (ch *channel) Accept() (Channel, <-chan *Request, error) { if ch.decided { return nil, nil, errDecidedAlready } - ch.maxIncomingPayload = channelMaxPacket confirm := channelOpenConfirmMsg{ PeersID: ch.remoteId, MyID: ch.localId, diff --git a/ssh/mux.go b/ssh/mux.go index 5775881..955560b 100644 --- a/ssh/mux.go +++ b/ssh/mux.go @@ -32,18 +32,21 @@ type chanList struct { offset uint32 } -// Assigns a channel ID to the given channel. -func (c *chanList) add(ch *channel) uint32 { +// add stores the given channel and assigns its localId while holding the +// lock, so that getChan can never return a channel whose localId is not yet +// initialized. +func (c *chanList) add(ch *channel) { c.Lock() defer c.Unlock() for i := range c.chans { if c.chans[i] == nil { c.chans[i] = ch - return uint32(i) + c.offset + ch.localId = uint32(i) + c.offset + return } } c.chans = append(c.chans, ch) - return uint32(len(c.chans)-1) + c.offset + ch.localId = uint32(len(c.chans)-1) + c.offset } // getChan returns the channel for the given ID. @@ -343,8 +346,6 @@ func (m *mux) OpenChannel(chanType string, extra []byte) (Channel, <-chan *Reque func (m *mux) openChannel(chanType string, extra []byte) (*channel, error) { ch := m.newChannel(chanType, channelOutbound, extra) - ch.maxIncomingPayload = channelMaxPacket - open := channelOpenMsg{ ChanType: chanType, PeersWindow: ch.myWindow, diff --git a/ssh/mux_test.go b/ssh/mux_test.go index 1370091..5196b33 100644 --- a/ssh/mux_test.go +++ b/ssh/mux_test.go @@ -1225,6 +1225,53 @@ func TestChannelCloseIdempotent(t *testing.T) { ch.close() // must be a no-op, not a panic } +func TestNewChannelFullyInitializedWhenPublished(t *testing.T) { + a, b := memPipe() + defer a.Close() + defer b.Close() + + m := newMux(a) + + const channels = 100 + stop := make(chan struct{}) + readerDone := make(chan struct{}) + go func() { + // Mimic the mux loop: fetch channels by id and read the fields it + // uses to route and validate packets. + defer close(readerDone) + for { + select { + case <-stop: + return + default: + } + for id := uint32(0); id < channels; id++ { + ch := m.chanList.getChan(id) + if ch == nil { + continue + } + if got := ch.localId; got != id { + t.Errorf("channel registered under id %d has localId %d", id, got) + return + } + if got := ch.maxIncomingPayload; got != channelMaxPacket { + t.Errorf("channel %d: maxIncomingPayload = %d, want %d", id, got, channelMaxPacket) + return + } + } + } + }() + + for range channels { + ch := m.newChannel("session", channelOutbound, nil) + if m.chanList.getChan(ch.localId) != ch { + t.Fatalf("channel %d not registered under its localId", ch.localId) + } + } + close(stop) + <-readerDone +} + func TestChannelConcurrentRequests(t *testing.T) { writer, reader, mux := channelPair(t) defer writer.Close()