filter: support filtering dump/flush by L3 protocol family (v4/v6)

Signed-off-by: Timo Beckers <[email protected]>

Author: ti-mo

conn.go

@@ -193,8 +193,8 @@ func (c *Conn) Dump(opts *DumpOptions) ([]Flow, error) {
 	return unmarshalFlows(nlm)
 }
 
-// DumpFilter gets all Conntrack connections from the kernel in the form of a list
-// of Flow objects, but only returns Flows matching the connmark specified in the Filter parameter.
+// DumpFilter gets all Conntrack connections from the kernel in the form of a
+// list of Flow objects. Only Flows matching the provided [Filter] are returned.
 func (c *Conn) DumpFilter(filter Filter, opts *DumpOptions) ([]Flow, error) {
 	if filter == nil {
 		return nil, fmt.Errorf("filter is nil")
@@ -209,7 +209,7 @@ func (c *Conn) DumpFilter(filter Filter, opts *DumpOptions) ([]Flow, error) {
 		netfilter.Header{
 			SubsystemID: netfilter.NFSubsysCTNetlink,
 			MessageType: netfilter.MessageType(msgType),
-			Family:      netfilter.ProtoUnspec, // ProtoUnspec dumps both IPv4 and IPv6
+			Family:      filter.family(),
 			Flags:       netlink.Request | netlink.Dump,
 		},
 		filter.marshal())
@@ -274,8 +274,8 @@ func (c *Conn) Flush() error {
 	return nil
 }
 
-// FlushFilter deletes all entries from the Conntrack table matching a given Filter.
-// Both IPv4 and IPv6 entries are considered for deletion.
+// FlushFilter deletes all entries from the Conntrack table matching a given
+// [Filter].
 func (c *Conn) FlushFilter(filter Filter) error {
 	if filter == nil {
 		return fmt.Errorf("filter is nil")
@@ -285,8 +285,16 @@ func (c *Conn) FlushFilter(filter Filter) error {
 		netfilter.Header{
 			SubsystemID: netfilter.NFSubsysCTNetlink,
 			MessageType: netfilter.MessageType(ctDelete),
-			Family:      netfilter.ProtoUnspec, // Family is ignored for flush
+			Family:      filter.family(),
 			Flags:       netlink.Request | netlink.Acknowledge,
+
+			// Request 'new' filtered flush behaviour as described in e7600865db32
+			// ("netfilter: ctnetlink: Fix regression in conntrack entry deletion").
+			// Before this patch, the kernel would ignore the nfgenmsg family and
+			// flush the entire table. As of 1ef7f50ccc6e ("netfilter: ctnetlink:
+			// support CTA_FILTER for flush"), the same can be accomplished by setting
+			// an empty CTA_FILTER.
+			Version: 1,
 		},
 		filter.marshal())
 

filter.go

@@ -11,6 +11,15 @@ import (
 //
 // Pass a filter to [Conn.DumpFilter] or [Conn.FlushFilter].
 type Filter interface {
+	// Family sets the address (L3) family to filter on, similar to conntrack's
+	// -f/--family.
+	//
+	// Common values are [netfilter.ProtoIPv4] and [netfilter.ProtoIPv6].
+	//
+	// Requires Linux 4.20 or later for [Conn.DumpFilter] and Linux 5.3 for
+	// [Conn.FlushFilter].
+	Family(l3 netfilter.ProtoFamily) Filter
+
 	// Mark sets the connmark to filter on, similar to conntrack's --mark option.
 	//
 	// When not specifying a mark mask, the kernel defaults to 0xFFFFFFFF, meaning
@@ -48,6 +57,8 @@ type Filter interface {
 	// Requires Linux 6.8 or later.
 	Zone(zone uint16) Filter
 
+	family() netfilter.ProtoFamily
+
 	marshal() []netfilter.Attribute
 }
 
@@ -58,6 +69,17 @@ func NewFilter() Filter {
 
 type filter struct {
 	f map[attributeType][]byte
+
+	l3 netfilter.ProtoFamily
+}
+
+func (f *filter) Family(l3 netfilter.ProtoFamily) Filter {
+	f.l3 = l3
+	return f
+}
+
+func (f *filter) family() netfilter.ProtoFamily {
+	return f.l3
 }
 
 func (f *filter) Mark(mark uint32) Filter {

flow_integration_test.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"github.com/ti-mo/netfilter"
 )
 
 // Create a given number of flows with a randomized component and check the amount
@@ -490,3 +491,35 @@ func TestStatusFilter(t *testing.T) {
 	require.NoError(t, err)
 	assert.Len(t, flows, 0)
 }
+
+func TestFamilyFilter(t *testing.T) {
+	c, _, err := makeNSConn()
+	require.NoError(t, err)
+
+	require.NoError(t, c.Create(NewFlow(unix.IPPROTO_TCP, 0, netip.MustParseAddr("1.2.3.4"), netip.MustParseAddr("5.6.7.8"), 1234, 80, 120, 0)))
+	require.NoError(t, c.Create(NewFlow(unix.IPPROTO_UDP, 0, netip.MustParseAddr("2a00:1450:400e:804::200e"), netip.MustParseAddr("2a00:1450:400e:804::200f"), 1234, 80, 120, 0)))
+
+	flows, err := c.Dump(nil)
+	require.NoError(t, err)
+	assert.Len(t, flows, 2, "expected 2 flows in total")
+
+	flows, err = c.DumpFilter(NewFilter().Family(netfilter.ProtoIPv4), nil)
+	require.NoError(t, err)
+	assert.Len(t, flows, 1)
+	assert.Equal(t, flows[0].TupleOrig.IP.SourceAddress, netip.MustParseAddr("1.2.3.4"))
+
+	flows, err = c.DumpFilter(NewFilter().Family(netfilter.ProtoIPv6), nil)
+	require.NoError(t, err)
+	assert.Len(t, flows, 1)
+	assert.Equal(t, flows[0].TupleOrig.IP.SourceAddress, netip.MustParseAddr("2a00:1450:400e:804::200e"))
+
+	assert.NoError(t, c.FlushFilter(NewFilter().Family(netfilter.ProtoIPv4)))
+	flows, err = c.Dump(nil)
+	require.NoError(t, err)
+	assert.Len(t, flows, 1, "expected 1 flow in total")
+
+	flows, err = c.DumpFilter(NewFilter().Family(netfilter.ProtoIPv6), nil)
+	require.NoError(t, err)
+	assert.Len(t, flows, 1)
+	assert.Equal(t, flows[0].TupleOrig.IP.SourceAddress, netip.MustParseAddr("2a00:1450:400e:804::200e"))
+}