Add methods in pkg/agent/route for NodeNetworkPolicy

pkg/agent/route/interfaces.go

@@ -18,6 +18,8 @@ import (
 	"net"
 	"time"
 
+	"k8s.io/apimachinery/pkg/util/sets"
+
 	"antrea.io/antrea/pkg/agent/config"
 	binding "antrea.io/antrea/pkg/ovs/openflow"
 )
@@ -90,4 +92,16 @@ type Interface interface {
 
 	// ClearConntrackEntryForService deletes a conntrack entry for a Service connection.
 	ClearConntrackEntryForService(svcIP net.IP, svcPort uint16, endpointIP net.IP, protocol binding.Protocol) error
+
+	// AddOrUpdateNodeNetworkPolicyIPSet adds or updates ipset created for NodeNetworkPolicy.
+	AddOrUpdateNodeNetworkPolicyIPSet(ipsetName string, prevIPSetEntries, curIPSetEntries sets.Set[string], isIPv6 bool) error
+
+	// DeleteNodeNetworkPolicyIPSet deletes ipset created for NodeNetworkPolicy.
+	DeleteNodeNetworkPolicyIPSet(ipsetName string, isIPv6 bool) error
+
+	// AddOrUpdateNodeNetworkPolicyIPTables adds or updates iptables chains and rules within the chains for NodeNetworkPolicy.
+	AddOrUpdateNodeNetworkPolicyIPTables(iptablesChains []string, iptablesRules [][]string, isIPv6 bool) error
+
+	// DeleteNodeNetworkPolicyIPTables deletes iptables chains and rules within the chains for NodeNetworkPolicy.
+	DeleteNodeNetworkPolicyIPTables(iptablesChains []string, isIPv6 bool) error
 }

pkg/agent/route/route_linux.go

@@ -126,6 +126,14 @@ type Client struct {
 	clusterNodeIP6s sync.Map
 	// The latest calculated Service CIDRs can be got from serviceCIDRProvider.
 	serviceCIDRProvider servicecidr.Interface
+	// nodeNetworkPolicyIPSetsIPv4 caches all existing IPv4 ipsets for NodeNetworkPolicy.
+	nodeNetworkPolicyIPSetsIPv4 sync.Map
+	// nodeNetworkPolicyIPSetsIPv6 caches all existing IPv6 ipsets for NodeNetworkPolicy.
+	nodeNetworkPolicyIPSetsIPv6 sync.Map
+	// nodeNetworkPolicyIPSetsIPv4 caches all existing IPv4 iptables chains and rules within the chains for NodeNetworkPolicy.
+	nodeNetworkPolicyIPTablesIPv4 sync.Map
+	// nodeNetworkPolicyIPSetsIPv6 caches all existing IPv6 iptables chains and rules within the chains for NodeNetworkPolicy.
+	nodeNetworkPolicyIPTablesIPv6 sync.Map
 }
 
 // NewClient returns a route client.
@@ -1700,3 +1708,85 @@ func generateNeigh(ip net.IP, linkIndex int) *netlink.Neigh {
 		HardwareAddr: globalVMAC,
 	}
 }
+
+func (c *Client) AddOrUpdateNodeNetworkPolicyIPSet(ipsetName string, prevIPSetEntries, curIPSetEntries sets.Set[string], isIPv6 bool) error {
+	ipsetEntriesToAdd := curIPSetEntries.Difference(prevIPSetEntries)
+	ipsetEntriesToDelete := prevIPSetEntries.Difference(curIPSetEntries)
+
+	if err := c.ipset.CreateIPSet(ipsetName, ipset.HashNet, isIPv6); err != nil {
+		return err
+	}
+	for ipsetEntry := range ipsetEntriesToAdd {
+		if err := c.ipset.AddEntry(ipsetName, ipsetEntry); err != nil {
+			return err
+		}
+	}
+	for ipsetEntry := range ipsetEntriesToDelete {
+		if err := c.ipset.DelEntry(ipsetName, ipsetEntry); err != nil {
+			return err
+		}
+	}
+	if isIPv6 {
+		c.nodeNetworkPolicyIPSetsIPv6.Store(ipsetName, curIPSetEntries)
+	} else {
+		c.nodeNetworkPolicyIPSetsIPv4.Store(ipsetName, curIPSetEntries)
+	}
+	return nil
+}
+
+func (c *Client) DeleteNodeNetworkPolicyIPSet(ipsetName string, isIPv6 bool) error {
+	if err := c.ipset.DestroyIPSet(ipsetName); err != nil {
+		return err
+	}
+	if isIPv6 {
+		c.nodeNetworkPolicyIPSetsIPv6.Delete(ipsetName)
+	} else {
+		c.nodeNetworkPolicyIPSetsIPv4.Delete(ipsetName)
+	}
+	return nil
+}
+
+func (c *Client) AddOrUpdateNodeNetworkPolicyIPTables(iptablesChains []string, iptablesRules [][]string, isIPv6 bool) error {
+	iptablesData := bytes.NewBuffer(nil)
+
+	writeLine(iptablesData, "*filter")
+	for _, iptablesChain := range iptablesChains {
+		writeLine(iptablesData, iptables.MakeChainLine(iptablesChain))
+	}
+	for _, rules := range iptablesRules {
+		for _, rule := range rules {
+			writeLine(iptablesData, rule)
+		}
+	}
+	writeLine(iptablesData, "COMMIT")
+
+	if err := c.iptables.Restore(iptablesData.String(), false, isIPv6); err != nil {
+		return err
+	}
+	for index, iptablesChain := range iptablesChains {
+		if isIPv6 {
+			c.nodeNetworkPolicyIPTablesIPv6.Store(iptablesChain, iptablesChains[index])
+		} else {
+			c.nodeNetworkPolicyIPTablesIPv4.Store(iptablesChain, iptablesChains[index])
+		}
+	}
+	return nil
+}
+
+func (c *Client) DeleteNodeNetworkPolicyIPTables(iptablesChains []string, isIPv6 bool) error {
+	ipProtocol := iptables.ProtocolIPv4
+	if isIPv6 {
+		ipProtocol = iptables.ProtocolIPv6
+	}
+	for _, iptablesChain := range iptablesChains {
+		if err := c.iptables.DeleteChain(ipProtocol, iptables.FilterTable, iptablesChain); err != nil {
+			return err
+		}
+		if isIPv6 {
+			c.nodeNetworkPolicyIPTablesIPv6.Delete(iptablesChain)
+		} else {
+			c.nodeNetworkPolicyIPTablesIPv4.Delete(iptablesChain)
+		}
+	}
+	return nil
+}

pkg/agent/route/route_linux_test.go

@@ -23,6 +23,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/vishvananda/netlink"
 	"go.uber.org/mock/gomock"
+	"k8s.io/apimachinery/pkg/util/sets"
 
 	"antrea.io/antrea/pkg/agent/config"
 	servicecidrtest "antrea.io/antrea/pkg/agent/servicecidr/testing"
@@ -1721,3 +1722,192 @@ func TestAddAndDeleteNodeIP(t *testing.T) {
 		})
 	}
 }
+
+func TestAddAndDeleteNodeNetworkPolicyIPSet(t *testing.T) {
+	ipv4SetName := "TEST-IPSET-4"
+	ipv4Net1 := "1.1.1.1/32"
+	ipv4Net2 := "2.2.2.2/32"
+	ipv4Net3 := "3.3.3.3/32"
+	ipv6SetName := "TEST-IPSET-6"
+	ipv6Net1 := "fec0::1111/128"
+	ipv6Net2 := "fec0::2222/128"
+	ipv6Net3 := "fec0::3333/128"
+
+	tests := []struct {
+		name             string
+		ipsetName        string
+		prevIPSetEntries sets.Set[string]
+		curIPSetEntries  sets.Set[string]
+		isIPv6           bool
+		expectedCalls    func(mockIPSet *ipsettest.MockInterfaceMockRecorder)
+	}{
+		{
+			name:            "IPv4, add an ipset and delete it",
+			ipsetName:       ipv4SetName,
+			curIPSetEntries: sets.New[string](ipv4Net1, ipv4Net3),
+			isIPv6:          false,
+			expectedCalls: func(mockIPSet *ipsettest.MockInterfaceMockRecorder) {
+				mockIPSet.CreateIPSet(ipv4SetName, ipset.HashNet, false).Times(1)
+				mockIPSet.AddEntry(ipv4SetName, ipv4Net1).Times(1)
+				mockIPSet.AddEntry(ipv4SetName, ipv4Net3).Times(1)
+				mockIPSet.DestroyIPSet(ipv4SetName).Times(1)
+			},
+		},
+		{
+			name:             "IPv4, update an ipset and delete it",
+			ipsetName:        ipv4SetName,
+			prevIPSetEntries: sets.New[string](ipv4Net1, ipv4Net2),
+			curIPSetEntries:  sets.New[string](ipv4Net1, ipv4Net3),
+			isIPv6:           false,
+			expectedCalls: func(mockIPSet *ipsettest.MockInterfaceMockRecorder) {
+				mockIPSet.CreateIPSet(ipv4SetName, ipset.HashNet, false).Times(1)
+				mockIPSet.AddEntry(ipv4SetName, ipv4Net3).Times(1)
+				mockIPSet.DelEntry(ipv4SetName, ipv4Net2).Times(1)
+				mockIPSet.DestroyIPSet(ipv4SetName).Times(1)
+			},
+		},
+		{
+			name:            "IPv6, add an ipset and delete it",
+			ipsetName:       ipv6SetName,
+			curIPSetEntries: sets.New[string](ipv6Net1, ipv6Net3),
+			isIPv6:          true,
+			expectedCalls: func(mockIPSet *ipsettest.MockInterfaceMockRecorder) {
+				mockIPSet.CreateIPSet(ipv6SetName, ipset.HashNet, true).Times(1)
+				mockIPSet.AddEntry(ipv6SetName, ipv6Net1).Times(1)
+				mockIPSet.AddEntry(ipv6SetName, ipv6Net3).Times(1)
+				mockIPSet.DestroyIPSet(ipv6SetName).Times(1)
+			},
+		},
+		{
+			name:             "IPv6, update an ipset and delete it",
+			ipsetName:        ipv6SetName,
+			prevIPSetEntries: sets.New[string](ipv6Net1, ipv6Net2),
+			curIPSetEntries:  sets.New[string](ipv6Net1, ipv6Net3),
+			isIPv6:           true,
+			expectedCalls: func(mockIPSet *ipsettest.MockInterfaceMockRecorder) {
+				mockIPSet.CreateIPSet(ipv6SetName, ipset.HashNet, true).Times(1)
+				mockIPSet.AddEntry(ipv6SetName, ipv6Net3).Times(1)
+				mockIPSet.DelEntry(ipv6SetName, ipv6Net2).Times(1)
+				mockIPSet.DestroyIPSet(ipv6SetName).Times(1)
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			ctrl := gomock.NewController(t)
+			mockIPSet := ipsettest.NewMockInterface(ctrl)
+			c := &Client{ipset: mockIPSet}
+			tt.expectedCalls(mockIPSet.EXPECT())
+
+			assert.NoError(t, c.AddOrUpdateNodeNetworkPolicyIPSet(tt.ipsetName, tt.prevIPSetEntries, tt.curIPSetEntries, tt.isIPv6))
+			var exists bool
+			if tt.isIPv6 {
+				_, exists = c.nodeNetworkPolicyIPSetsIPv6.Load(tt.ipsetName)
+			} else {
+				_, exists = c.nodeNetworkPolicyIPSetsIPv4.Load(tt.ipsetName)
+			}
+			assert.True(t, exists)
+
+			assert.NoError(t, c.DeleteNodeNetworkPolicyIPSet(tt.ipsetName, tt.isIPv6))
+			if tt.isIPv6 {
+				_, exists = c.nodeNetworkPolicyIPSetsIPv6.Load(tt.ipsetName)
+			} else {
+				_, exists = c.nodeNetworkPolicyIPSetsIPv4.Load(tt.ipsetName)
+			}
+			assert.False(t, exists)
+		})
+	}
+}
+
+func TestAddAndDeleteNodeNetworkPolicyIPTables(t *testing.T) {
+	chain1 := "TEST-CHAIN1"
+	chain2 := "TEST-CHAIN2"
+	chains := []string{chain1, chain2}
+	rules := [][]string{
+		{
+			"-A TEST-CHAIN1 -j ACCEPT -p tcp --dport 80",
+			"-A TEST-CHAIN1 -j ACCEPT -p tcp --dport 443",
+		},
+		{
+			"-A TEST-CHAIN2 -j ACCEPT -p tcp --dport 80",
+			"-A TEST-CHAIN2 -j ACCEPT -p tcp --dport 443",
+		},
+	}
+	tests := []struct {
+		name           string
+		iptablesChains []string
+		iptablesRules  [][]string
+		isIPv6         bool
+		expectedCalls  func(mockIPTables *iptablestest.MockInterfaceMockRecorder)
+	}{
+		{
+			name:           "IPv4",
+			iptablesChains: chains,
+			iptablesRules:  rules,
+			isIPv6:         false,
+			expectedCalls: func(mockIPTables *iptablestest.MockInterfaceMockRecorder) {
+				mockIPTables.Restore(`*filter
+:TEST-CHAIN1 - [0:0]
+:TEST-CHAIN2 - [0:0]
+-A TEST-CHAIN1 -j ACCEPT -p tcp --dport 80
+-A TEST-CHAIN1 -j ACCEPT -p tcp --dport 443
+-A TEST-CHAIN2 -j ACCEPT -p tcp --dport 80
+-A TEST-CHAIN2 -j ACCEPT -p tcp --dport 443
+COMMIT
+`, false, false)
+				mockIPTables.DeleteChain(iptables.ProtocolIPv4, iptables.FilterTable, chain1)
+				mockIPTables.DeleteChain(iptables.ProtocolIPv4, iptables.FilterTable, chain2)
+			},
+		},
+		{
+			name:           "IPv6",
+			iptablesChains: chains,
+			iptablesRules:  rules,
+			isIPv6:         true,
+			expectedCalls: func(mockIPTables *iptablestest.MockInterfaceMockRecorder) {
+				mockIPTables.Restore(`*filter
+:TEST-CHAIN1 - [0:0]
+:TEST-CHAIN2 - [0:0]
+-A TEST-CHAIN1 -j ACCEPT -p tcp --dport 80
+-A TEST-CHAIN1 -j ACCEPT -p tcp --dport 443
+-A TEST-CHAIN2 -j ACCEPT -p tcp --dport 80
+-A TEST-CHAIN2 -j ACCEPT -p tcp --dport 443
+COMMIT
+`, false, true)
+				mockIPTables.DeleteChain(iptables.ProtocolIPv6, iptables.FilterTable, chain1)
+				mockIPTables.DeleteChain(iptables.ProtocolIPv6, iptables.FilterTable, chain2)
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			ctrl := gomock.NewController(t)
+			mockIPTables := iptablestest.NewMockInterface(ctrl)
+			c := &Client{iptables: mockIPTables}
+			tt.expectedCalls(mockIPTables.EXPECT())
+
+			assert.NoError(t, c.AddOrUpdateNodeNetworkPolicyIPTables(tt.iptablesChains, tt.iptablesRules, tt.isIPv6))
+			for _, chain := range chains {
+				var exists bool
+				if tt.isIPv6 {
+					_, exists = c.nodeNetworkPolicyIPTablesIPv6.Load(chain)
+				} else {
+					_, exists = c.nodeNetworkPolicyIPTablesIPv4.Load(chain)
+				}
+				assert.True(t, exists)
+			}
+
+			assert.NoError(t, c.DeleteNodeNetworkPolicyIPTables(tt.iptablesChains, tt.isIPv6))
+			for _, chain := range chains {
+				var exists bool
+				if tt.isIPv6 {
+					_, exists = c.nodeNetworkPolicyIPTablesIPv6.Load(chain)
+				} else {
+					_, exists = c.nodeNetworkPolicyIPTablesIPv4.Load(chain)
+				}
+				assert.False(t, exists)
+			}
+		})
+	}
+}

pkg/agent/route/route_windows.go

@@ -573,3 +573,19 @@ func (c *Client) DeleteRouteForLink(dstCIDR *net.IPNet, linkIndex int) error {
 func (c *Client) ClearConntrackEntryForService(svcIP net.IP, svcPort uint16, endpointIP net.IP, protocol binding.Protocol) error {
 	return errors.New("ClearConntrackEntryForService is not implemented on Windows")
 }
+
+func (c *Client) AddOrUpdateNodeNetworkPolicyIPSet(ipsetName string, prevIPSetEntries, curIPSetEntries sets.Set[string], isIPv6 bool) error {
+	return errors.New("AddOrUpdateNodeNetworkPolicyIPSet is not implemented on Windows")
+}
+
+func (c *Client) DeleteNodeNetworkPolicyIPSet(ipsetName string, isIPv6 bool) error {
+	return errors.New("DeleteNodeNetworkPolicyIPSet is not implemented on Windows")
+}
+
+func (c *Client) AddOrUpdateNodeNetworkPolicyIPTables(iptablesChains []string, iptablesRules [][]string, isIPv6 bool) error {
+	return errors.New("AddOrUpdateNodeNetworkPolicyIPTables is not implemented on Windows")
+}
+
+func (c *Client) DeleteNodeNetworkPolicyIPTables(iptablesChains []string, isIPv6 bool) error {
+	return errors.New("DeleteNodeNetworkPolicyIPTables is not implemented on Windows")
+}