From a629c1271e12be7da01b82fb726394562bed67f2 Mon Sep 17 00:00:00 2001 From: Gregor Noczinski Date: Sun, 27 Sep 2026 12:38:08 +0200 Subject: [PATCH 1/2] Allow custom listeners for reverse TCP forwards --- tcpip.go | 43 ++++++- tcpip_test.go | 337 +++++++++++++++++++++++++++++++++++++++++++++++++- 2 files changed, 376 insertions(+), 4 deletions(-) diff --git a/tcpip.go b/tcpip.go index 810201c..2bd65d0 100644 --- a/tcpip.go +++ b/tcpip.go @@ -114,6 +114,14 @@ type remoteForwardChannelData struct { type ForwardedTCPHandler struct { Logger log.Logger + // ListenCallback creates the listener after reverse forwarding has been + // authorized and a forward slot reserved. It receives the original bind + // address and port (including port 0). Any returned listener, even on + // error, is closed by the handler; Close must unblock Accept. + // If nil, the handler listens on TCP in the server process. Callbacks must + // honor context cancellation. + ListenCallback func(ctx Context, conn gossh.ConnMetadata, bindHost string, bindPort uint32) (net.Listener, error) + forwards map[forwardKey]*forward sync.Mutex } @@ -210,14 +218,43 @@ func (h *ForwardedTCPHandler) HandleSSHRequest(response RequestResponseWriter, r forwardLimiter.release() return response.Reject(nil) } - ln, err := new(net.ListenConfig).Listen(ctx, "tcp", addr) + var ln net.Listener + if h.ListenCallback != nil { + ln, err = h.ListenCallback(ctx, conn, reqPayload.BindAddr, reqPayload.BindPort) + } else { + ln, err = new(net.ListenConfig).Listen(ctx, "tcp", addr) + } if err != nil { + closeQuietly(ln) finishAcquisition(nil) forwardLimiter.release() return locateError(ErrorScopeForwarding, ErrorOperationListen, fmt.Errorf("listen for tcpip-forward on %s: %w", addr, err)) } - _, destPortStr, _ := net.SplitHostPort(ln.Addr().String()) - destPort, _ := strconv.ParseUint(destPortStr, 10, 16) + if ln == nil { + finishAcquisition(nil) + forwardLimiter.release() + return locateError(ErrorScopeForwarding, ErrorOperationListen, fmt.Errorf("listen for tcpip-forward on %s: callback returned no listener", addr)) + } + var destPortStr string + var destPort uint64 + listenerAddr := ln.Addr() + if listenerAddr != nil { + _, destPortStr, err = net.SplitHostPort(listenerAddr.String()) + if err == nil { + destPort, err = strconv.ParseUint(destPortStr, 10, 16) + } + } else { + err = errors.New("listener returned no address") + } + if err == nil && destPort == 0 { + err = errors.New("listener has no bound port") + } + if err != nil { + closeQuietly(ln) + finishAcquisition(nil) + forwardLimiter.release() + return locateError(ErrorScopeForwarding, ErrorOperationListen, fmt.Errorf("listen for tcpip-forward on %s: invalid listener address %v: %w", addr, listenerAddr, err)) + } key := forwardKey{conn: conn, addr: net.JoinHostPort(reqPayload.BindAddr, destPortStr)} f := newForward(ln, forwardLimiter.release) unregisterForward := finishAcquisition(f.close) diff --git a/tcpip_test.go b/tcpip_test.go index 5d57a61..f2a9578 100644 --- a/tcpip_test.go +++ b/tcpip_test.go @@ -3,11 +3,13 @@ package ssh import ( "bytes" "context" + "errors" "io" "net" "strconv" "strings" "sync" + "sync/atomic" "testing" "time" @@ -170,8 +172,335 @@ func TestRemotePortZeroForwardIsRemovedOnCancel(t *testing.T) { }, time.Second, time.Millisecond) } -func TestRemoteForwardIsRemovedWhenRequestReplyFails(t *testing.T) { +type trackedTCPListener struct { + net.Listener + closed chan struct{} + once sync.Once +} + +func (l *trackedTCPListener) Close() error { + err := l.Listener.Close() + l.once.Do(func() { close(l.closed) }) + return err +} + +func TestRemoteForwardDefaultListener(t *testing.T) { handler := &ForwardedTCPHandler{} + _, client, cleanup := newTestSession(t, &Server{ + ReversePortForwardingCallback: func(Context, gossh.ConnMetadata, string, uint32) (bool, error) { return true, nil }, + RequestHandlers: map[string]RequestHandler{ + "tcpip-forward": handler.HandleSSHRequest, + "cancel-tcpip-forward": handler.HandleSSHRequest, + }, + }, nil) + defer cleanup() + + listener, err := client.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + defer closeQuietly(listener) + addr, ok := listener.Addr().(*net.TCPAddr) + require.True(t, ok) + require.NotZero(t, addr.Port) + peer, err := net.Dial("tcp", listener.Addr().String()) + require.NoError(t, err) + defer closeQuietly(peer) + forwarded := acceptStreamLocal(t, listener) + defer closeQuietly(forwarded) + assertStreamLocalExchange(t, forwarded, peer) + require.NoError(t, listener.Close()) + require.Eventually(t, func() bool { + handler.Lock() + defer handler.Unlock() + return len(handler.forwards) == 0 + }, time.Second, time.Millisecond) +} + +func TestRemoteForwardListenCallbackPortAndMetadata(t *testing.T) { + type call struct { + host string + port uint32 + user string + sameCtx bool + } + calls := make(chan call, 1) + var authorized atomic.Bool + var serverListener *trackedTCPListener + handler := &ForwardedTCPHandler{ + ListenCallback: func(ctx Context, conn gossh.ConnMetadata, host string, port uint32) (net.Listener, error) { + if !authorized.Load() { + return nil, errors.New("listener called before authorization") + } + ln, err := new(net.ListenConfig).Listen(ctx, "tcp", "127.0.0.1:0") + if err != nil { + return nil, err + } + serverListener = &trackedTCPListener{Listener: ln, closed: make(chan struct{})} + calls <- call{host, port, conn.User(), ctx.Value(ContextKeyConn) == conn} + return serverListener, nil + }, + } + _, client, cleanup := newTestSession(t, &Server{ + ReversePortForwardingCallback: func(Context, gossh.ConnMetadata, string, uint32) (bool, error) { + authorized.Store(true) + return true, nil + }, + RequestHandlers: map[string]RequestHandler{ + "tcpip-forward": handler.HandleSSHRequest, + "cancel-tcpip-forward": handler.HandleSSHRequest, + }, + }, nil) + defer cleanup() + + incoming := client.HandleChannelOpen(forwardedTCPChannelType) + require.NotNil(t, incoming) + const bindHost = "localhost" + ok, reply, err := client.SendRequest("tcpip-forward", true, gossh.Marshal(&remoteForwardRequest{BindAddr: bindHost, BindPort: 0})) + require.NoError(t, err) + require.True(t, ok) + require.Equal(t, call{bindHost, 0, "testuser", true}, <-calls) + var success remoteForwardSuccess + require.NoError(t, gossh.Unmarshal(reply, &success)) + require.Equal(t, uint32(serverListener.Addr().(*net.TCPAddr).Port), success.BindPort) + require.NotZero(t, success.BindPort) + peer, err := net.Dial("tcp", serverListener.Addr().String()) + require.NoError(t, err) + defer closeQuietly(peer) + var opened gossh.NewChannel + select { + case opened = <-incoming: + require.NotNil(t, opened) + case <-time.After(time.Second): + t.Fatal("forwarded-tcpip channel was not opened") + } + var data remoteForwardChannelData + require.NoError(t, gossh.Unmarshal(opened.ExtraData(), &data)) + peerHost, peerPort, err := net.SplitHostPort(peer.LocalAddr().String()) + require.NoError(t, err) + parsedPeerPort, err := strconv.Atoi(peerPort) + require.NoError(t, err) + require.Equal(t, remoteForwardChannelData{DestAddr: bindHost, DestPort: success.BindPort, OriginAddr: peerHost, OriginPort: uint32(parsedPeerPort)}, data) + channel, requests, err := opened.Accept() + require.NoError(t, err) + defer closeQuietly(channel) + go gossh.DiscardRequests(requests) + + ok, _, err = client.SendRequest("cancel-tcpip-forward", true, gossh.Marshal(&remoteForwardCancelRequest{BindAddr: bindHost, BindPort: success.BindPort})) + require.NoError(t, err) + require.True(t, ok) + select { + case <-serverListener.closed: + case <-time.After(time.Second): + t.Fatal("cancel did not close callback listener") + } + require.Eventually(t, func() bool { + handler.Lock() + defer handler.Unlock() + return len(handler.forwards) == 0 + }, time.Second, time.Millisecond) +} + +func TestRemoteForwardListenCallbackUsesBoundPortInsteadOfRequestedPort(t *testing.T) { + listeners := make(chan *trackedTCPListener, 1) + handler := &ForwardedTCPHandler{ + ListenCallback: func(ctx Context, _ gossh.ConnMetadata, _ string, _ uint32) (net.Listener, error) { + ln, err := new(net.ListenConfig).Listen(ctx, "tcp", "127.0.0.1:0") + if err != nil { + return nil, err + } + tracked := &trackedTCPListener{Listener: ln, closed: make(chan struct{})} + listeners <- tracked + return tracked, nil + }, + } + _, client, cleanup := newTestSession(t, &Server{ + ReversePortForwardingCallback: func(Context, gossh.ConnMetadata, string, uint32) (bool, error) { return true, nil }, + RequestHandlers: map[string]RequestHandler{ + "tcpip-forward": handler.HandleSSHRequest, + "cancel-tcpip-forward": handler.HandleSSHRequest, + }, + }, nil) + defer cleanup() + + const requestedPort = 1 + ok, reply, err := client.SendRequest("tcpip-forward", true, gossh.Marshal(&remoteForwardRequest{BindAddr: "127.0.0.1", BindPort: requestedPort})) + require.NoError(t, err) + require.True(t, ok) + listener := <-listeners + var success remoteForwardSuccess + require.NoError(t, gossh.Unmarshal(reply, &success)) + require.Equal(t, uint32(listener.Addr().(*net.TCPAddr).Port), success.BindPort) + require.NotEqual(t, uint32(requestedPort), success.BindPort) + ok, _, err = client.SendRequest("cancel-tcpip-forward", true, gossh.Marshal(&remoteForwardCancelRequest{BindAddr: "127.0.0.1", BindPort: success.BindPort})) + require.NoError(t, err) + require.True(t, ok) + select { + case <-listener.closed: + case <-time.After(time.Second): + t.Fatal("cancel did not close callback listener") + } +} + +func TestRemoteForwardListenCallbackFailureReleasesLimit(t *testing.T) { + for _, failure := range []string{"error", "listener and error", "nil listener", "invalid address"} { + t.Run(failure, func(t *testing.T) { + maxForwards := 1 + var attempts atomic.Int32 + var failedListener *trackedTCPListener + handler := &ForwardedTCPHandler{ + ListenCallback: func(ctx Context, _ gossh.ConnMetadata, _ string, _ uint32) (net.Listener, error) { + if attempts.Add(1) != 1 { + return new(net.ListenConfig).Listen(ctx, "tcp", "127.0.0.1:0") + } + if failure == "error" { + return nil, errors.New("bind failed") + } + if failure == "nil listener" { + return nil, nil + } + ln, err := new(net.ListenConfig).Listen(ctx, "tcp", "127.0.0.1:0") + if err != nil { + return nil, err + } + failedListener = &trackedTCPListener{Listener: ln, closed: make(chan struct{})} + if failure == "listener and error" { + return failedListener, errors.New("bind failed") + } + return invalidTCPListener{failedListener}, nil + }, + } + _, client, cleanup := newTestSession(t, &Server{ + MaxReverseForwardsPerConnection: &maxForwards, + ReversePortForwardingCallback: func(Context, gossh.ConnMetadata, string, uint32) (bool, error) { return true, nil }, + RequestHandlers: map[string]RequestHandler{ + "tcpip-forward": handler.HandleSSHRequest, + "cancel-tcpip-forward": handler.HandleSSHRequest, + }, + }, nil) + defer cleanup() + + _, err := client.Listen("tcp", "127.0.0.1:0") + require.Error(t, err) + handler.Lock() + count := len(handler.forwards) + handler.Unlock() + require.Zero(t, count) + if failedListener != nil { + select { + case <-failedListener.closed: + case <-time.After(time.Second): + t.Fatal("failed callback listener was not closed") + } + } + listener, err := client.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err, "failed bind must release the forward slot") + require.NoError(t, listener.Close()) + require.Equal(t, int32(2), attempts.Load()) + }) + } +} + +type invalidTCPListener struct{ *trackedTCPListener } + +func (invalidTCPListener) Addr() net.Addr { return &net.UnixAddr{Name: "invalid", Net: "unix"} } + +func TestRemoteForwardListenCallbackRequiresAuthorizationAndCapacity(t *testing.T) { + maxForwards := 1 + var calls atomic.Int32 + handler := &ForwardedTCPHandler{ + ListenCallback: func(ctx Context, _ gossh.ConnMetadata, _ string, _ uint32) (net.Listener, error) { + calls.Add(1) + return new(net.ListenConfig).Listen(ctx, "tcp", "127.0.0.1:0") + }, + } + _, client, cleanup := newTestSession(t, &Server{ + MaxReverseForwardsPerConnection: &maxForwards, + ReversePortForwardingCallback: func(_ Context, _ gossh.ConnMetadata, _ string, port uint32) (bool, error) { + return port != 12345, nil + }, + RequestHandlers: map[string]RequestHandler{ + "tcpip-forward": handler.HandleSSHRequest, + "cancel-tcpip-forward": handler.HandleSSHRequest, + }, + }, nil) + defer cleanup() + + _, err := client.Listen("tcp", "127.0.0.1:12345") + require.Error(t, err) + require.Zero(t, calls.Load()) + first, err := client.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + defer closeQuietly(first) + _, err = client.Listen("tcp", "127.0.0.1:0") + require.Error(t, err) + require.Equal(t, int32(1), calls.Load()) +} + +func TestRemoteForwardListenCallbackClosesOnDisconnect(t *testing.T) { + listeners := make(chan *trackedTCPListener, 1) + disconnected := make(chan bool, 1) + handler := &ForwardedTCPHandler{ + ListenCallback: func(ctx Context, _ gossh.ConnMetadata, _ string, _ uint32) (net.Listener, error) { + ln, err := new(net.ListenConfig).Listen(ctx, "tcp", "127.0.0.1:0") + if err != nil { + return nil, err + } + tracked := &trackedTCPListener{Listener: ln, closed: make(chan struct{})} + listeners <- tracked + return tracked, nil + }, + } + _, client, cleanup := newTestSession(t, &Server{ + ReversePortForwardingCallback: func(Context, gossh.ConnMetadata, string, uint32) (bool, error) { return true, nil }, + DisconnectCallback: func(Context, net.Conn) error { + select { + case listener := <-listeners: + select { + case <-listener.closed: + disconnected <- true + default: + disconnected <- false + } + default: + disconnected <- false + } + return nil + }, + RequestHandlers: map[string]RequestHandler{ + "tcpip-forward": handler.HandleSSHRequest, + "cancel-tcpip-forward": handler.HandleSSHRequest, + }, + }, nil) + defer cleanup() + + listener, err := client.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + defer closeQuietly(listener) + require.NoError(t, client.Close()) + select { + case closed := <-disconnected: + require.True(t, closed) + case <-time.After(time.Second): + t.Fatal("disconnect callback was not called") + } + handler.Lock() + count := len(handler.forwards) + handler.Unlock() + require.Zero(t, count) +} + +func TestRemoteForwardIsRemovedWhenRequestReplyFails(t *testing.T) { + listeners := make(chan *trackedTCPListener, 1) + handler := &ForwardedTCPHandler{ + ListenCallback: func(ctx Context, _ gossh.ConnMetadata, _ string, _ uint32) (net.Listener, error) { + ln, err := new(net.ListenConfig).Listen(ctx, "tcp", "127.0.0.1:0") + if err != nil { + return nil, err + } + tracked := &trackedTCPListener{Listener: ln, closed: make(chan struct{})} + listeners <- tracked + return tracked, nil + }, + } session, client, cleanup := newTestSession(t, &Server{ Handler: func(Session) error { return nil }, ReversePortForwardingCallback: func(Context, gossh.ConnMetadata, string, uint32) (bool, error) { return true, nil }, @@ -192,6 +521,12 @@ func TestRemoteForwardIsRemovedWhenRequestReplyFails(t *testing.T) { _, _, err := client.SendRequest("tcpip-forward", true, gossh.Marshal(&remoteForwardRequest{BindAddr: "127.0.0.1", BindPort: 0})) require.Error(t, err) + listener := <-listeners + select { + case <-listener.closed: + case <-time.After(time.Second): + t.Fatal("failed reply did not close callback listener") + } require.Eventually(t, func() bool { handler.Lock() From cf68f8a8d36040b1ac2a85cea190cb30c8202806 Mon Sep 17 00:00:00 2001 From: Gregor Noczinski Date: Sun, 27 Sep 2026 13:00:50 +0200 Subject: [PATCH 2/2] Prevent reverse TCP forward listener key collisions --- tcpip.go | 22 ++++++- tcpip_test.go | 175 ++++++++++++++++++++++++++++++++++++++++++++++---- 2 files changed, 182 insertions(+), 15 deletions(-) diff --git a/tcpip.go b/tcpip.go index 2bd65d0..f56cf2e 100644 --- a/tcpip.go +++ b/tcpip.go @@ -118,6 +118,7 @@ type ForwardedTCPHandler struct { // authorized and a forward slot reserved. It receives the original bind // address and port (including port 0). Any returned listener, even on // error, is closed by the handler; Close must unblock Accept. + // For a nonzero requested port, the listener must bind that same port. // If nil, the handler listens on TCP in the server process. Callbacks must // honor context cancellation. ListenCallback func(ctx Context, conn gossh.ConnMetadata, bindHost string, bindPort uint32) (net.Listener, error) @@ -255,10 +256,22 @@ func (h *ForwardedTCPHandler) HandleSSHRequest(response RequestResponseWriter, r forwardLimiter.release() return locateError(ErrorScopeForwarding, ErrorOperationListen, fmt.Errorf("listen for tcpip-forward on %s: invalid listener address %v: %w", addr, listenerAddr, err)) } - key := forwardKey{conn: conn, addr: net.JoinHostPort(reqPayload.BindAddr, destPortStr)} + if reqPayload.BindPort != 0 && uint64(reqPayload.BindPort) != destPort { + closeQuietly(ln) + finishAcquisition(nil) + forwardLimiter.release() + return locateError(ErrorScopeForwarding, ErrorOperationListen, fmt.Errorf("listen for tcpip-forward on %s: listener bound unexpected port %d", addr, destPort)) + } + key := forwardKey{conn: conn, addr: net.JoinHostPort(reqPayload.BindAddr, strconv.FormatUint(destPort, 10))} f := newForward(ln, forwardLimiter.release) unregisterForward := finishAcquisition(f.close) h.Lock() + if _, exists := h.forwards[key]; exists { + h.Unlock() + unregisterForward() + f.close() + return response.Reject(nil) + } h.forwards[key] = f h.Unlock() responseState := request.response @@ -388,6 +401,13 @@ func (h *ForwardedTCPHandler) HandleSSHRequest(response RequestResponseWriter, r h.Unlock() if ok { f.close() + // Keep old channel opens from reaching a new forward with this key. + f.waitForPendingOpens() + h.Lock() + if h.forwards[key] == f { + delete(h.forwards, key) + } + h.Unlock() return response.Accept(nil) } return response.Reject(nil) diff --git a/tcpip_test.go b/tcpip_test.go index f2a9578..c36acfa 100644 --- a/tcpip_test.go +++ b/tcpip_test.go @@ -299,8 +299,8 @@ func TestRemoteForwardListenCallbackPortAndMetadata(t *testing.T) { }, time.Second, time.Millisecond) } -func TestRemoteForwardListenCallbackUsesBoundPortInsteadOfRequestedPort(t *testing.T) { - listeners := make(chan *trackedTCPListener, 1) +func TestRemoteForwardListenCallbackRejectsUnexpectedPort(t *testing.T) { + listeners := make(chan *trackedTCPListener, 2) handler := &ForwardedTCPHandler{ ListenCallback: func(ctx Context, _ gossh.ConnMetadata, _ string, _ uint32) (net.Listener, error) { ln, err := new(net.ListenConfig).Listen(ctx, "tcp", "127.0.0.1:0") @@ -312,8 +312,10 @@ func TestRemoteForwardListenCallbackUsesBoundPortInsteadOfRequestedPort(t *testi return tracked, nil }, } + maxForwards := 1 _, client, cleanup := newTestSession(t, &Server{ - ReversePortForwardingCallback: func(Context, gossh.ConnMetadata, string, uint32) (bool, error) { return true, nil }, + MaxReverseForwardsPerConnection: &maxForwards, + ReversePortForwardingCallback: func(Context, gossh.ConnMetadata, string, uint32) (bool, error) { return true, nil }, RequestHandlers: map[string]RequestHandler{ "tcpip-forward": handler.HandleSSHRequest, "cancel-tcpip-forward": handler.HandleSSHRequest, @@ -322,22 +324,167 @@ func TestRemoteForwardListenCallbackUsesBoundPortInsteadOfRequestedPort(t *testi defer cleanup() const requestedPort = 1 - ok, reply, err := client.SendRequest("tcpip-forward", true, gossh.Marshal(&remoteForwardRequest{BindAddr: "127.0.0.1", BindPort: requestedPort})) - require.NoError(t, err) - require.True(t, ok) + _, err := client.Listen("tcp", "127.0.0.1:1") + require.Error(t, err) listener := <-listeners - var success remoteForwardSuccess - require.NoError(t, gossh.Unmarshal(reply, &success)) - require.Equal(t, uint32(listener.Addr().(*net.TCPAddr).Port), success.BindPort) - require.NotEqual(t, uint32(requestedPort), success.BindPort) - ok, _, err = client.SendRequest("cancel-tcpip-forward", true, gossh.Marshal(&remoteForwardCancelRequest{BindAddr: "127.0.0.1", BindPort: success.BindPort})) - require.NoError(t, err) - require.True(t, ok) + require.NotEqual(t, requestedPort, listener.Addr().(*net.TCPAddr).Port) select { case <-listener.closed: case <-time.After(time.Second): - t.Fatal("cancel did not close callback listener") + t.Fatal("rejected listener was not closed") + } + handler.Lock() + count := len(handler.forwards) + handler.Unlock() + require.Zero(t, count) + next, err := client.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err, "rejected listener must release the forward slot") + require.NoError(t, next.Close()) +} + +type formattedTCPAddr string + +func (formattedTCPAddr) Network() string { return "tcp" } +func (a formattedTCPAddr) String() string { return string(a) } + +type formattedTCPListener struct { + *trackedTCPListener + addr net.Addr +} + +func (l *formattedTCPListener) Addr() net.Addr { return l.addr } + +func TestRemoteForwardListenCallbackCanonicalizesPort(t *testing.T) { + listeners := make(chan *trackedTCPListener, 1) + handler := &ForwardedTCPHandler{ + ListenCallback: func(ctx Context, _ gossh.ConnMetadata, _ string, _ uint32) (net.Listener, error) { + ln, err := new(net.ListenConfig).Listen(ctx, "tcp", "127.0.0.1:0") + if err != nil { + return nil, err + } + tracked := &trackedTCPListener{Listener: ln, closed: make(chan struct{})} + listeners <- tracked + port := strconv.Itoa(ln.Addr().(*net.TCPAddr).Port) + return &formattedTCPListener{tracked, formattedTCPAddr(net.JoinHostPort("127.0.0.1", "000"+port))}, nil + }, + } + _, client, cleanup := newTestSession(t, &Server{ + ReversePortForwardingCallback: func(Context, gossh.ConnMetadata, string, uint32) (bool, error) { return true, nil }, + RequestHandlers: map[string]RequestHandler{ + "tcpip-forward": handler.HandleSSHRequest, + "cancel-tcpip-forward": handler.HandleSSHRequest, + }, + }, nil) + defer cleanup() + + listener, err := client.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + tracked := <-listeners + require.Equal(t, tracked.Addr().(*net.TCPAddr).Port, listener.Addr().(*net.TCPAddr).Port) + require.NoError(t, listener.Close()) + select { + case <-tracked.closed: + case <-time.After(time.Second): + t.Fatal("cancel did not close listener with formatted port") } + require.Eventually(t, func() bool { + handler.Lock() + defer handler.Unlock() + return len(handler.forwards) == 0 + }, time.Second, time.Millisecond) +} + +func TestRemoteForwardListenCallbackRejectsDuplicateBoundPort(t *testing.T) { + maxForwards := 2 + listeners := make(chan *trackedTCPListener, 3) + var firstPort atomic.Int32 + var calls atomic.Int32 + handler := &ForwardedTCPHandler{ + ListenCallback: func(ctx Context, _ gossh.ConnMetadata, _ string, _ uint32) (net.Listener, error) { + ln, err := new(net.ListenConfig).Listen(ctx, "tcp", "127.0.0.1:0") + if err != nil { + return nil, err + } + tracked := &trackedTCPListener{Listener: ln, closed: make(chan struct{})} + listeners <- tracked + switch calls.Add(1) { + case 1: + firstPort.Store(int32(ln.Addr().(*net.TCPAddr).Port)) + case 2: + return &formattedTCPListener{tracked, &net.TCPAddr{IP: net.IPv4(127, 0, 0, 2), Port: int(firstPort.Load())}}, nil + } + return tracked, nil + }, + } + _, client, cleanup := newTestSession(t, &Server{ + MaxReverseForwardsPerConnection: &maxForwards, + ReversePortForwardingCallback: func(Context, gossh.ConnMetadata, string, uint32) (bool, error) { return true, nil }, + RequestHandlers: map[string]RequestHandler{ + "tcpip-forward": handler.HandleSSHRequest, + "cancel-tcpip-forward": handler.HandleSSHRequest, + }, + }, nil) + defer cleanup() + + first, err := client.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + defer closeQuietly(first) + firstServer := <-listeners + _, err = client.Listen("tcp", "127.0.0.1:0") + require.Error(t, err) + duplicate := <-listeners + select { + case <-duplicate.closed: + case <-time.After(time.Second): + t.Fatal("duplicate listener was not closed") + } + select { + case <-firstServer.closed: + t.Fatal("duplicate closed the existing listener") + default: + } + + third, err := client.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err, "duplicate must release its forward slot") + defer closeQuietly(third) + <-listeners + peer, err := net.Dial("tcp", firstServer.Addr().String()) + require.NoError(t, err) + defer closeQuietly(peer) + forwarded := acceptStreamLocal(t, first) + defer closeQuietly(forwarded) + assertStreamLocalExchange(t, forwarded, peer) + require.NoError(t, first.Close()) + require.NoError(t, third.Close()) + require.Eventually(t, func() bool { + handler.Lock() + defer handler.Unlock() + return len(handler.forwards) == 0 + }, time.Second, time.Millisecond) +} + +func TestRemoteForwardListenCallbackCanRebindAfterCancel(t *testing.T) { + handler := &ForwardedTCPHandler{ + ListenCallback: func(ctx Context, _ gossh.ConnMetadata, host string, port uint32) (net.Listener, error) { + return new(net.ListenConfig).Listen(ctx, "tcp", net.JoinHostPort(host, strconv.Itoa(int(port)))) + }, + } + _, client, cleanup := newTestSession(t, &Server{ + ReversePortForwardingCallback: func(Context, gossh.ConnMetadata, string, uint32) (bool, error) { return true, nil }, + RequestHandlers: map[string]RequestHandler{ + "tcpip-forward": handler.HandleSSHRequest, + "cancel-tcpip-forward": handler.HandleSSHRequest, + }, + }, nil) + defer cleanup() + + first, err := client.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + addr := first.Addr().String() + require.NoError(t, first.Close()) + second, err := client.Listen("tcp", addr) + require.NoError(t, err, "a confirmed cancel must allow immediate reuse of the same port") + require.NoError(t, second.Close()) } func TestRemoteForwardListenCallbackFailureReleasesLimit(t *testing.T) {