From 9c234abeaa2132290671f975cce59e97ee962b1a Mon Sep 17 00:00:00 2001 From: "Francisco A. Lozano" Date: Sun, 1 Nov 2026 08:26:13 +0100 Subject: [PATCH] fix(transfer): fix passive port multiplexing races A reservation's Close deleted whatever reservation the port held for that IP, so a transfer ending after its data connection was dispatched removed the next session's reservation. Only delete the reservation being closed. deliver could send to connCh after Close drained it, leaving the connection with nobody to read or close it. Decide under stateMu, which Close takes before draining. Sessions from different IPs using a port for the first time raced to bind it, and the losers moved on to the next port or got a 421. Hold the lock across the bind. released was only ever written, so closing replaces it. Co-Authored-By: Claude Opus 5.5 --- passive_multiplexer.go | 67 +++++++-------------- passive_multiplexer_test.go | 117 ++++++++++++++++++++++++++++++++++++ 2 files changed, 140 insertions(+), 44 deletions(-) diff --git a/passive_multiplexer.go b/passive_multiplexer.go index 2d770ad9..7ca0bc75 100644 --- a/passive_multiplexer.go +++ b/passive_multiplexer.go @@ -85,38 +85,20 @@ func (m *passiveListenersManager) close() error { func (m *passiveListenersManager) getOrCreate(port int) (*sharedPassiveListener, error) { m.mu.Lock() - if m.closed { - m.mu.Unlock() + defer m.mu.Unlock() + if m.closed { return nil, net.ErrClosed } if listener, ok := m.listeners[port]; ok { - m.mu.Unlock() - return listener, nil } - m.mu.Unlock() listener, err := newSharedPassiveListener(port, m.logger.With("passivePort", port)) if err != nil { return nil, err } - m.mu.Lock() - defer m.mu.Unlock() - - if m.closed { - _ = listener.close() - - return nil, net.ErrClosed - } - - if existing, ok := m.listeners[port]; ok { - _ = listener.close() - - return existing, nil - } - m.listeners[port] = listener return listener, nil @@ -225,13 +207,12 @@ func (l *sharedPassiveListener) dispatch(conn net.Conn) { } } -func (l *sharedPassiveListener) release(remoteIP string) { +func (l *sharedPassiveListener) release(reservation *passiveReservationListener) { l.mu.Lock() defer l.mu.Unlock() - if reservation, ok := l.reservations[remoteIP]; ok { - delete(l.reservations, remoteIP) - reservation.markReleased() + if l.reservations[reservation.remoteIP] == reservation { + delete(l.reservations, reservation.remoteIP) } } @@ -271,7 +252,7 @@ type passiveReservationListener struct { closeOnce sync.Once stateMu sync.Mutex deadline time.Time - released bool + closing bool failureErr error } @@ -307,8 +288,8 @@ func (l *passiveReservationListener) Accept() (net.Conn, error) { func (l *passiveReservationListener) Close() error { l.closeOnce.Do(func() { - l.parent.release(l.remoteIP) - l.markReleased() + l.parent.release(l) + l.markClosing() select { case conn := <-l.connCh: @@ -337,21 +318,26 @@ func (l *passiveReservationListener) SetDeadline(deadline time.Time) error { return nil } +// deliver hands conn over unless Close started: connCh has room for the single connection a reservation +// gets, so the send cannot block, and Close drains connCh only after setting closing. func (l *passiveReservationListener) deliver(conn net.Conn) bool { - select { - case <-l.closedCh: + l.stateMu.Lock() + defer l.stateMu.Unlock() + + if l.closing { return false - default: } - select { - case l.connCh <- conn: - l.markReleased() + l.connCh <- conn - return true - case <-l.closedCh: - return false - } + return true +} + +func (l *passiveReservationListener) markClosing() { + l.stateMu.Lock() + defer l.stateMu.Unlock() + + l.closing = true } func (l *passiveReservationListener) fail(err error) { @@ -364,13 +350,6 @@ func (l *passiveReservationListener) fail(err error) { _ = l.Close() } -func (l *passiveReservationListener) markReleased() { - l.stateMu.Lock() - defer l.stateMu.Unlock() - - l.released = true -} - func (l *passiveReservationListener) getDeadline() time.Time { l.stateMu.Lock() defer l.stateMu.Unlock() diff --git a/passive_multiplexer_test.go b/passive_multiplexer_test.go index cb375fe5..3f9a2da8 100644 --- a/passive_multiplexer_test.go +++ b/passive_multiplexer_test.go @@ -1,11 +1,14 @@ package ftpserver import ( + "context" "errors" "fmt" "io" "log/slog" "net" + "sync" + "sync/atomic" "testing" "time" @@ -18,6 +21,11 @@ type trackingTestConn struct { closed bool } +type closeCountingConn struct { + testNetConn + closed atomic.Bool +} + const testServerListenAddr = "127.0.0.1:0" var ( @@ -32,6 +40,12 @@ func (c *trackingTestConn) Close() error { return nil } +func (c *closeCountingConn) Close() error { + c.closed.Store(true) + + return nil +} + func getFreePassivePort(t *testing.T) int { t.Helper() @@ -107,6 +121,109 @@ func TestPassiveListenersManagerRejectsSameIPForSamePort(t *testing.T) { req.Nil(deadlineSetter) } +func TestPassiveReservationCloseKeepsNewerReservation(t *testing.T) { + port := getFreePassivePort(t) + manager := newPassiveListenersManager(slog.New(slog.DiscardHandler)) + defer func() { + require.NoError(t, manager.close()) + }() + + portRange := PortRange{Start: port, End: port} + clientIP := net.ParseIP("127.0.0.1") + + _, first, _, err := manager.reserve(clientIP, portRange) + require.NoError(t, err) + + dialer := net.Dialer{Timeout: time.Second} + firstClient, err := dialer.DialContext(context.Background(), "tcp", fmt.Sprintf("127.0.0.1:%d", port)) + require.NoError(t, err) + defer func() { _ = firstClient.Close() }() + + firstConn, err := first.Accept() + require.NoError(t, err) + defer func() { _ = firstConn.Close() }() + + // The first data connection consumed its reservation, so the port is free for the same IP + _, second, secondDeadline, err := manager.reserve(clientIP, portRange) + require.NoError(t, err) + defer func() { _ = second.Close() }() + + require.NoError(t, first.Close()) + + secondClient, err := dialer.DialContext(context.Background(), "tcp", fmt.Sprintf("127.0.0.1:%d", port)) + require.NoError(t, err) + defer func() { _ = secondClient.Close() }() + + require.NoError(t, secondDeadline.SetDeadline(time.Now().Add(5*time.Second))) + secondConn, err := second.Accept() + require.NoError(t, err) + require.NoError(t, secondConn.Close()) +} + +func TestPassiveReservationDeliverCloseRace(t *testing.T) { + const rounds = 50000 + + parent := &sharedPassiveListener{ + logger: slog.New(slog.DiscardHandler), + reservations: make(map[string]*passiveReservationListener), + } + clientIP := net.ParseIP("127.0.0.1") + + for range rounds { + reservation, err := parent.reserve(clientIP) + require.NoError(t, err) + delete(parent.reservations, clientIP.String()) // as dispatch does before calling deliver + + conn := &closeCountingConn{} + start := make(chan struct{}) + var delivered bool + var waitGroup sync.WaitGroup + + waitGroup.Go(func() { + <-start + delivered = reservation.deliver(conn) + }) + waitGroup.Go(func() { + <-start + _ = reservation.Close() + }) + close(start) + waitGroup.Wait() + + if delivered { + require.True(t, conn.closed.Load(), "the connection was left with nobody to read or close it") + } + require.Empty(t, reservation.connCh) + } +} + +func TestPassiveListenersManagerConcurrentFirstUse(t *testing.T) { + const ( + rounds = 50 + clients = 8 + ) + + for range rounds { + port := getFreePassivePort(t) + manager := newPassiveListenersManager(slog.New(slog.DiscardHandler)) + portRange := PortRange{Start: port, End: port} + errs := make([]error, clients) + + var waitGroup sync.WaitGroup + for client := range clients { + waitGroup.Go(func() { + _, _, _, errs[client] = manager.reserve(net.ParseIP(fmt.Sprintf("127.0.0.%d", client+1)), portRange) + }) + } + waitGroup.Wait() + + for _, err := range errs { + require.NoError(t, err) + } + require.NoError(t, manager.close()) + } +} + func TestPassiveListenersManagerCloseReleasesReservation(t *testing.T) { req := require.New(t) port := getFreePassivePort(t)