From 23abb5743c5a3f86b4e92d36488747dcd33f7cdb Mon Sep 17 00:00:00 2001 From: Viktor Liu Date: Thu, 12 Feb 2026 20:11:12 +0800 Subject: [PATCH] Treated tombstoned conns as new --- client/firewall/uspfilter/conntrack/tcp.go | 2 +- .../firewall/uspfilter/conntrack/tcp_test.go | 151 ++++++++++++++++++ client/firewall/uspfilter/log/log.go | 15 +- 3 files changed, 165 insertions(+), 3 deletions(-) diff --git a/client/firewall/uspfilter/conntrack/tcp.go b/client/firewall/uspfilter/conntrack/tcp.go index 8d64412e0..992c10769 100644 --- a/client/firewall/uspfilter/conntrack/tcp.go +++ b/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 } diff --git a/client/firewall/uspfilter/conntrack/tcp_test.go b/client/firewall/uspfilter/conntrack/tcp_test.go index bb440f70a..52bc3858f 100644 --- a/client/firewall/uspfilter/conntrack/tcp_test.go +++ b/client/firewall/uspfilter/conntrack/tcp_test.go @@ -485,6 +485,157 @@ func TestTCPAbnormalSequences(t *testing.T) { }) } +// TestTCPPortReuseTombstone verifies that a new connection on a port with a +// tombstoned (closed) conntrack entry is properly tracked. Without the fix, +// updateIfExists treats tombstoned entries as live, causing track() to skip +// creating a new connection. The subsequent SYN-ACK then fails IsValidInbound +// because the entry is tombstoned, and the response packet gets dropped by ACL. +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 diff --git a/client/firewall/uspfilter/log/log.go b/client/firewall/uspfilter/log/log.go index 66308defc..c6ca55e70 100644 --- a/client/firewall/uspfilter/log/log.go +++ b/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 {