[client] Fix port reuse in tcp conntrack

client/firewall/uspfilter/conntrack/tcp.go

@@ -115,6 +115,17 @@ func (t *TCPConnTrack) IsTombstone() bool {
 	return t.tombstone.Load()
 }
 
+// IsSupersededBy returns true if this connection should be replaced by a new one
+// carrying the given flags. Tombstoned connections are always superseded; TIME-WAIT
+// connections are superseded by a pure SYN (a new connection attempt for the same
+// four-tuple, as contemplated by RFC 1122 §4.2.2.13 and RFC 6191).
+func (t *TCPConnTrack) IsSupersededBy(flags uint8) bool {
+	if t.tombstone.Load() {
+		return true
+	}
+	return flags&TCPSyn != 0 && flags&TCPAck == 0 && TCPState(t.state.Load()) == TCPStateTimeWait
+}
+
 // SetTombstone safely marks the connection for deletion
 func (t *TCPConnTrack) SetTombstone() {
 	t.tombstone.Store(true)
@@ -169,7 +180,7 @@ func (t *TCPTracker) updateIfExists(srcIP, dstIP netip.Addr, srcPort, dstPort ui
 	conn, exists := t.connections[key]
 	t.mutex.RUnlock()
 
-	if exists {
+	if exists && !conn.IsSupersededBy(flags) {
 		t.updateState(key, conn, flags, direction, size)
 		return key, uint16(conn.DNATOrigPort.Load()), true
 	}
@@ -241,7 +252,7 @@ func (t *TCPTracker) IsValidInbound(srcIP, dstIP netip.Addr, srcPort, dstPort ui
 	conn, exists := t.connections[key]
 	t.mutex.RUnlock()
 
-	if !exists || conn.IsTombstone() {
+	if !exists || conn.IsSupersededBy(flags) {
 		return false
 	}
 

client/firewall/uspfilter/conntrack/tcp_test.go

@@ -485,6 +485,259 @@ func TestTCPAbnormalSequences(t *testing.T) {
 	})
 }
 
+// TestTCPPortReuseTombstone verifies that a new connection on a port with a
+// tombstoned (closed) conntrack entry is properly tracked. A SYN on a
+// tombstoned entry must replace it so that the subsequent SYN-ACK is accepted.
+func TestTCPPortReuseTombstone(t *testing.T) {
+	srcIP := netip.MustParseAddr("100.64.0.1")
+	dstIP := netip.MustParseAddr("100.64.0.2")
+	srcPort := uint16(12345)
+	dstPort := uint16(80)
+
+	t.Run("Outbound port reuse after graceful close", func(t *testing.T) {
+		tracker := NewTCPTracker(DefaultTCPTimeout, logger, flowLogger)
+		defer tracker.Close()
+
+		key := ConnKey{SrcIP: srcIP, DstIP: dstIP, SrcPort: srcPort, DstPort: dstPort}
+
+		// Establish and gracefully close a connection (server-initiated close)
+		establishConnection(t, tracker, srcIP, dstIP, srcPort, dstPort)
+
+		// Server sends FIN
+		valid := tracker.IsValidInbound(dstIP, srcIP, dstPort, srcPort, TCPFin|TCPAck, 0)
+		require.True(t, valid)
+
+		// Client sends FIN-ACK
+		tracker.TrackOutbound(srcIP, dstIP, srcPort, dstPort, TCPFin|TCPAck, 0)
+
+		// Server sends final ACK
+		valid = tracker.IsValidInbound(dstIP, srcIP, dstPort, srcPort, TCPAck, 0)
+		require.True(t, valid)
+
+		// Connection should be tombstoned
+		conn := tracker.connections[key]
+		require.NotNil(t, conn, "old connection should still be in map")
+		require.True(t, conn.IsTombstone(), "old connection should be tombstoned")
+
+		// Now reuse the same port for a new connection
+		tracker.TrackOutbound(srcIP, dstIP, srcPort, dstPort, TCPSyn, 100)
+
+		// The old tombstoned entry should be replaced with a new one
+		newConn := tracker.connections[key]
+		require.NotNil(t, newConn, "new connection should exist")
+		require.False(t, newConn.IsTombstone(), "new connection should not be tombstoned")
+		require.Equal(t, TCPStateSynSent, newConn.GetState())
+
+		// SYN-ACK for the new connection should be valid
+		valid = tracker.IsValidInbound(dstIP, srcIP, dstPort, srcPort, TCPSyn|TCPAck, 100)
+		require.True(t, valid, "SYN-ACK for new connection on reused port should be accepted")
+		require.Equal(t, TCPStateEstablished, newConn.GetState())
+
+		// Data transfer should work
+		tracker.TrackOutbound(srcIP, dstIP, srcPort, dstPort, TCPAck, 100)
+		valid = tracker.IsValidInbound(dstIP, srcIP, dstPort, srcPort, TCPPush|TCPAck, 500)
+		require.True(t, valid, "data should be allowed on new connection")
+	})
+
+	t.Run("Outbound port reuse after RST", func(t *testing.T) {
+		tracker := NewTCPTracker(DefaultTCPTimeout, logger, flowLogger)
+		defer tracker.Close()
+
+		key := ConnKey{SrcIP: srcIP, DstIP: dstIP, SrcPort: srcPort, DstPort: dstPort}
+
+		// Establish and RST a connection
+		establishConnection(t, tracker, srcIP, dstIP, srcPort, dstPort)
+		valid := tracker.IsValidInbound(dstIP, srcIP, dstPort, srcPort, TCPRst|TCPAck, 0)
+		require.True(t, valid)
+
+		conn := tracker.connections[key]
+		require.True(t, conn.IsTombstone(), "RST connection should be tombstoned")
+
+		// Reuse the same port
+		tracker.TrackOutbound(srcIP, dstIP, srcPort, dstPort, TCPSyn, 100)
+
+		newConn := tracker.connections[key]
+		require.NotNil(t, newConn)
+		require.False(t, newConn.IsTombstone())
+		require.Equal(t, TCPStateSynSent, newConn.GetState())
+
+		valid = tracker.IsValidInbound(dstIP, srcIP, dstPort, srcPort, TCPSyn|TCPAck, 100)
+		require.True(t, valid, "SYN-ACK should be accepted after RST tombstone")
+	})
+
+	t.Run("Inbound port reuse after close", func(t *testing.T) {
+		tracker := NewTCPTracker(DefaultTCPTimeout, logger, flowLogger)
+		defer tracker.Close()
+
+		clientIP := srcIP
+		serverIP := dstIP
+		clientPort := srcPort
+		serverPort := dstPort
+		key := ConnKey{SrcIP: clientIP, DstIP: serverIP, SrcPort: clientPort, DstPort: serverPort}
+
+		// Inbound connection: client SYN → server SYN-ACK → client ACK
+		tracker.TrackInbound(clientIP, serverIP, clientPort, serverPort, TCPSyn, nil, 100, 0)
+		tracker.TrackOutbound(serverIP, clientIP, serverPort, clientPort, TCPSyn|TCPAck, 100)
+		tracker.TrackInbound(clientIP, serverIP, clientPort, serverPort, TCPAck, nil, 100, 0)
+
+		conn := tracker.connections[key]
+		require.Equal(t, TCPStateEstablished, conn.GetState())
+
+		// Server-initiated close to reach Closed/tombstoned:
+		// Server FIN (opposite dir) → CloseWait
+		tracker.TrackOutbound(serverIP, clientIP, serverPort, clientPort, TCPFin|TCPAck, 100)
+		require.Equal(t, TCPStateCloseWait, conn.GetState())
+		// Client FIN-ACK (same dir as conn) → LastAck
+		tracker.TrackInbound(clientIP, serverIP, clientPort, serverPort, TCPFin|TCPAck, nil, 100, 0)
+		require.Equal(t, TCPStateLastAck, conn.GetState())
+		// Server final ACK (opposite dir) → Closed → tombstoned
+		tracker.TrackOutbound(serverIP, clientIP, serverPort, clientPort, TCPAck, 100)
+
+		require.True(t, conn.IsTombstone())
+
+		// New inbound connection on same ports
+		tracker.TrackInbound(clientIP, serverIP, clientPort, serverPort, TCPSyn, nil, 100, 0)
+
+		newConn := tracker.connections[key]
+		require.NotNil(t, newConn)
+		require.False(t, newConn.IsTombstone())
+		require.Equal(t, TCPStateSynReceived, newConn.GetState())
+
+		// Complete handshake: server SYN-ACK, then client ACK
+		tracker.TrackOutbound(serverIP, clientIP, serverPort, clientPort, TCPSyn|TCPAck, 100)
+		tracker.TrackInbound(clientIP, serverIP, clientPort, serverPort, TCPAck, nil, 100, 0)
+		require.Equal(t, TCPStateEstablished, newConn.GetState())
+	})
+
+	t.Run("Late ACK on tombstoned connection is harmless", func(t *testing.T) {
+		tracker := NewTCPTracker(DefaultTCPTimeout, logger, flowLogger)
+		defer tracker.Close()
+
+		key := ConnKey{SrcIP: srcIP, DstIP: dstIP, SrcPort: srcPort, DstPort: dstPort}
+
+		// Establish and close via passive close (server-initiated FIN → Closed → tombstoned)
+		establishConnection(t, tracker, srcIP, dstIP, srcPort, dstPort)
+		tracker.IsValidInbound(dstIP, srcIP, dstPort, srcPort, TCPFin|TCPAck, 0) // CloseWait
+		tracker.TrackOutbound(srcIP, dstIP, srcPort, dstPort, TCPFin|TCPAck, 0)  // LastAck
+		tracker.IsValidInbound(dstIP, srcIP, dstPort, srcPort, TCPAck, 0)        // Closed
+
+		conn := tracker.connections[key]
+		require.True(t, conn.IsTombstone())
+
+		// Late ACK should be rejected (tombstoned)
+		valid := tracker.IsValidInbound(dstIP, srcIP, dstPort, srcPort, TCPAck, 0)
+		require.False(t, valid, "late ACK on tombstoned connection should be rejected")
+
+		// Late outbound ACK should not create a new connection (not a SYN)
+		tracker.TrackOutbound(srcIP, dstIP, srcPort, dstPort, TCPAck, 0)
+		require.True(t, tracker.connections[key].IsTombstone(), "late outbound ACK should not replace tombstoned entry")
+	})
+}
+
+func TestTCPPortReuseTimeWait(t *testing.T) {
+	srcIP := netip.MustParseAddr("100.64.0.1")
+	dstIP := netip.MustParseAddr("100.64.0.2")
+	srcPort := uint16(12345)
+	dstPort := uint16(80)
+
+	t.Run("Outbound port reuse during TIME-WAIT (active close)", func(t *testing.T) {
+		tracker := NewTCPTracker(DefaultTCPTimeout, logger, flowLogger)
+		defer tracker.Close()
+
+		key := ConnKey{SrcIP: srcIP, DstIP: dstIP, SrcPort: srcPort, DstPort: dstPort}
+
+		// Establish connection
+		establishConnection(t, tracker, srcIP, dstIP, srcPort, dstPort)
+
+		// Active close: client (outbound initiator) sends FIN first
+		tracker.TrackOutbound(srcIP, dstIP, srcPort, dstPort, TCPFin|TCPAck, 0)
+		conn := tracker.connections[key]
+		require.Equal(t, TCPStateFinWait1, conn.GetState())
+
+		// Server ACKs the FIN
+		valid := tracker.IsValidInbound(dstIP, srcIP, dstPort, srcPort, TCPAck, 0)
+		require.True(t, valid)
+		require.Equal(t, TCPStateFinWait2, conn.GetState())
+
+		// Server sends its own FIN
+		valid = tracker.IsValidInbound(dstIP, srcIP, dstPort, srcPort, TCPFin|TCPAck, 0)
+		require.True(t, valid)
+		require.Equal(t, TCPStateTimeWait, conn.GetState())
+
+		// Client sends final ACK (TIME-WAIT stays, not tombstoned)
+		tracker.TrackOutbound(srcIP, dstIP, srcPort, dstPort, TCPAck, 0)
+		require.False(t, conn.IsTombstone(), "TIME-WAIT should not be tombstoned")
+
+		// New outbound SYN on the same port (port reuse during TIME-WAIT)
+		tracker.TrackOutbound(srcIP, dstIP, srcPort, dstPort, TCPSyn, 100)
+
+		// Per RFC 1122/6191, new SYN during TIME-WAIT should start a new connection
+		newConn := tracker.connections[key]
+		require.NotNil(t, newConn, "new connection should exist")
+		require.False(t, newConn.IsTombstone(), "new connection should not be tombstoned")
+		require.Equal(t, TCPStateSynSent, newConn.GetState(), "new connection should be in SYN-SENT")
+
+		// SYN-ACK for new connection should be valid
+		valid = tracker.IsValidInbound(dstIP, srcIP, dstPort, srcPort, TCPSyn|TCPAck, 100)
+		require.True(t, valid, "SYN-ACK for new connection should be accepted")
+		require.Equal(t, TCPStateEstablished, newConn.GetState())
+	})
+
+	t.Run("Inbound SYN during TIME-WAIT falls through to normal tracking", func(t *testing.T) {
+		tracker := NewTCPTracker(DefaultTCPTimeout, logger, flowLogger)
+		defer tracker.Close()
+
+		key := ConnKey{SrcIP: srcIP, DstIP: dstIP, SrcPort: srcPort, DstPort: dstPort}
+
+		// Establish outbound connection and close via active close → TIME-WAIT
+		establishConnection(t, tracker, srcIP, dstIP, srcPort, dstPort)
+		tracker.TrackOutbound(srcIP, dstIP, srcPort, dstPort, TCPFin|TCPAck, 0)
+		tracker.IsValidInbound(dstIP, srcIP, dstPort, srcPort, TCPAck, 0)
+		tracker.IsValidInbound(dstIP, srcIP, dstPort, srcPort, TCPFin|TCPAck, 0)
+		tracker.TrackOutbound(srcIP, dstIP, srcPort, dstPort, TCPAck, 0)
+
+		conn := tracker.connections[key]
+		require.Equal(t, TCPStateTimeWait, conn.GetState())
+
+		// Inbound SYN on same ports during TIME-WAIT: IsValidInbound returns false
+		// so the filter falls through to ACL check + TrackInbound (which creates
+		// a new connection via track() → updateIfExists skips TIME-WAIT for SYN)
+		valid := tracker.IsValidInbound(dstIP, srcIP, dstPort, srcPort, TCPSyn, 0)
+		require.False(t, valid, "inbound SYN during TIME-WAIT should fail conntrack validation")
+
+		// Simulate what the filter does next: TrackInbound via the normal path
+		tracker.TrackInbound(dstIP, srcIP, dstPort, srcPort, TCPSyn, nil, 100, 0)
+
+		// The new inbound connection uses the inverted key (dst→src becomes src→dst in track)
+		invertedKey := ConnKey{SrcIP: dstIP, DstIP: srcIP, SrcPort: dstPort, DstPort: srcPort}
+		newConn := tracker.connections[invertedKey]
+		require.NotNil(t, newConn, "new inbound connection should be tracked")
+		require.Equal(t, TCPStateSynReceived, newConn.GetState())
+		require.False(t, newConn.IsTombstone())
+	})
+
+	t.Run("Late retransmit during TIME-WAIT still allowed", func(t *testing.T) {
+		tracker := NewTCPTracker(DefaultTCPTimeout, logger, flowLogger)
+		defer tracker.Close()
+
+		key := ConnKey{SrcIP: srcIP, DstIP: dstIP, SrcPort: srcPort, DstPort: dstPort}
+
+		// Establish and active close → TIME-WAIT
+		establishConnection(t, tracker, srcIP, dstIP, srcPort, dstPort)
+		tracker.TrackOutbound(srcIP, dstIP, srcPort, dstPort, TCPFin|TCPAck, 0)
+		tracker.IsValidInbound(dstIP, srcIP, dstPort, srcPort, TCPAck, 0)
+		tracker.IsValidInbound(dstIP, srcIP, dstPort, srcPort, TCPFin|TCPAck, 0)
+		tracker.TrackOutbound(srcIP, dstIP, srcPort, dstPort, TCPAck, 0)
+
+		conn := tracker.connections[key]
+		require.Equal(t, TCPStateTimeWait, conn.GetState())
+
+		// Late ACK retransmits during TIME-WAIT should still be accepted
+		valid := tracker.IsValidInbound(dstIP, srcIP, dstPort, srcPort, TCPAck, 0)
+		require.True(t, valid, "retransmitted ACK during TIME-WAIT should be accepted")
+	})
+}
+
 func TestTCPTimeoutHandling(t *testing.T) {
 	// Create tracker with a very short timeout for testing
 	shortTimeout := 100 * time.Millisecond

client/firewall/uspfilter/log/log.go

@@ -5,6 +5,8 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"os"
+	"strconv"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -16,9 +18,18 @@ const (
 	maxBatchSize         = 1024 * 16
 	maxMessageSize       = 1024 * 2
 	defaultFlushInterval = 2 * time.Second
-	logChannelSize       = 1000
+	defaultLogChanSize   = 1000
 )
 
+func getLogChannelSize() int {
+	if v := os.Getenv("NB_USPFILTER_LOG_BUFFER"); v != "" {
+		if n, err := strconv.Atoi(v); err == nil && n > 0 {
+			return n
+		}
+	}
+	return defaultLogChanSize
+}
+
 type Level uint32
 
 const (
@@ -69,7 +80,7 @@ type Logger struct {
 func NewFromLogrus(logrusLogger *log.Logger) *Logger {
 	l := &Logger{
 		output:     logrusLogger.Out,
-		msgChannel: make(chan logMessage, logChannelSize),
+		msgChannel: make(chan logMessage, getLogChannelSize()),
 		shutdown:   make(chan struct{}),
 		bufPool: sync.Pool{
 			New: func() any {