Treated tombstoned conns as new

Author: lixmal

client/firewall/uspfilter/conntrack/tcp.go

@@ -169,7 +169,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.IsTombstone() {
 		t.updateState(key, conn, flags, direction, size)
 		return key, uint16(conn.DNATOrigPort.Load()), true
 	}

client/firewall/uspfilter/conntrack/tcp_test.go

@@ -485,6 +485,155 @@ 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 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 {