Add iptables interface for implementing Egress

pkg/agent/route/interfaces.go

@@ -45,6 +45,12 @@ type Interface interface {
 	// if linkName is nil, it should remove the routes.
 	UnMigrateRoutesFromGw(route *net.IPNet, linkName string) error
 
+	// AddSNATRule should add rule to SNAT outgoing traffic with the mark, using the provided SNAT IP.
+	AddSNATRule(snatIP net.IP, mark uint32) error
+
+	// DeleteSNATRule should delete rule to SNAT outgoing traffic with the mark.
+	DeleteSNATRule(mark uint32) error
+
 	// Run starts the sync loop.
 	Run(stopCh <-chan struct{})
 }

pkg/agent/route/route_linux.go

@@ -80,6 +80,8 @@ type Client struct {
 	nodeRoutes sync.Map
 	// nodeNeighbors caches IPv6 Neighbors to remote host gateway
 	nodeNeighbors sync.Map
+	// markToSNATIP caches marks to SNAT IPs. It's used in Egress feature.
+	markToSNATIP sync.Map
 	// iptablesInitialized is used to notify when iptables initialization is done.
 	iptablesInitialized chan struct{}
 }
@@ -239,9 +241,21 @@ func (c *Client) syncIPTables() error {
 		}
 	}
 
+	snatMarkToIPv4 := map[uint32]net.IP{}
+	snatMarkToIPv6 := map[uint32]net.IP{}
+	c.markToSNATIP.Range(func(key, value interface{}) bool {
+		snatMark := key.(uint32)
+		snatIP := value.(net.IP)
+		if snatIP.To4() != nil {
+			snatMarkToIPv4[snatMark] = snatIP
+		} else {
+			snatMarkToIPv6[snatMark] = snatIP
+		}
+		return true
+	})
 	// Use iptables-restore to configure IPv4 settings.
 	if v4Enabled {
-		iptablesData := c.restoreIptablesData(c.nodeConfig.PodIPv4CIDR, antreaPodIPSet)
+		iptablesData := c.restoreIptablesData(c.nodeConfig.PodIPv4CIDR, antreaPodIPSet, snatMarkToIPv4)
 		// Setting --noflush to keep the previous contents (i.e. non antrea managed chains) of the tables.
 		if err := c.ipt.Restore(iptablesData.Bytes(), false, false); err != nil {
 			return err
@@ -250,7 +264,7 @@ func (c *Client) syncIPTables() error {
 
 	// Use ip6tables-restore to configure IPv6 settings.
 	if v6Enabled {
-		iptablesData := c.restoreIptablesData(c.nodeConfig.PodIPv6CIDR, antreaPodIP6Set)
+		iptablesData := c.restoreIptablesData(c.nodeConfig.PodIPv6CIDR, antreaPodIP6Set, snatMarkToIPv6)
 		// Setting --noflush to keep the previous contents (i.e. non antrea managed chains) of the tables.
 		if err := c.ipt.Restore(iptablesData.Bytes(), false, true); err != nil {
 			return err
@@ -259,7 +273,7 @@ func (c *Client) syncIPTables() error {
 	return nil
 }
 
-func (c *Client) restoreIptablesData(podCIDR *net.IPNet, podIPSet string) *bytes.Buffer {
+func (c *Client) restoreIptablesData(podCIDR *net.IPNet, podIPSet string, snatMarkToIP map[uint32]net.IP) *bytes.Buffer {
 	// Create required rules in the antrea chains.
 	// Use iptables-restore as it flushes the involved chains and creates the desired rules
 	// with a single call, instead of string matching to clean up stale rules.
@@ -340,6 +354,17 @@ func (c *Client) restoreIptablesData(podCIDR *net.IPNet, podIPSet string) *bytes
 
 	writeLine(iptablesData, "*nat")
 	writeLine(iptablesData, iptables.MakeChainLine(antreaPostRoutingChain))
+	// Egress rule must be inserted before the default masquerade rule.
+	for snatMark, snatIP := range snatMarkToIP {
+		// Cannot reuse snatRuleSpec to generate the rule as it doesn't have "`" in the comment.
+		writeLine(iptablesData, []string{
+			"-A", antreaPostRoutingChain,
+			"-m", "comment", "--comment", `"Antrea: snat pod to external packets"`,
+			"-m", "mark", "--mark", fmt.Sprintf("%#08x/%#08x", snatMark, types.SNATIPMarkMask),
+			"-j", iptables.SNATTarget, "--to", snatIP.String(),
+		}...)
+	}
+
 	if !c.noSNAT {
 		writeLine(iptablesData, []string{
 			"-A", antreaPostRoutingChain,
@@ -657,3 +682,31 @@ func (c *Client) UnMigrateRoutesFromGw(route *net.IPNet, linkName string) error
 	}
 	return nil
 }
+
+func snatRuleSpec(snatIP net.IP, snatMark uint32) []string {
+	return []string{
+		"-m", "comment", "--comment", "Antrea: snat pod to external packets",
+		"-m", "mark", "--mark", fmt.Sprintf("%#08x/%#08x", snatMark, types.SNATIPMarkMask),
+		"-j", iptables.SNATTarget, "--to", snatIP.String(),
+	}
+}
+
+func (c *Client) AddSNATRule(snatIP net.IP, mark uint32) error {
+	protocol := iptables.ProtocolIPv4
+	if snatIP.To4() == nil {
+		protocol = iptables.ProtocolIPv6
+	}
+	c.markToSNATIP.Store(mark, snatIP)
+	return c.ipt.InsertRule(protocol, iptables.NATTable, antreaPostRoutingChain, snatRuleSpec(snatIP, mark))
+}
+
+func (c *Client) DeleteSNATRule(mark uint32) error {
+	value, ok := c.markToSNATIP.Load(mark)
+	if !ok {
+		klog.Warningf("Didn't find SNAT rule with mark %#x", mark)
+		return nil
+	}
+	c.markToSNATIP.Delete(mark)
+	snatIP := value.(net.IP)
+	return c.ipt.DeleteRule(iptables.NATTable, antreaPostRoutingChain, snatRuleSpec(snatIP, mark))
+}

pkg/agent/route/route_windows.go

@@ -206,3 +206,11 @@ func (c *Client) initFwRules() error {
 	}
 	return nil
 }
+
+func (c *Client) AddSNATRule(snatIP net.IP, mark uint32) error {
+	return nil
+}
+
+func (c *Client) DeleteSNATRule(mark uint32) error {
+	return nil
+}

pkg/agent/util/iptables/iptables.go

@@ -39,6 +39,7 @@ const (
 	MarkTarget       = "MARK"
 	ConnTrackTarget  = "CT"
 	NoTrackTarget    = "NOTRACK"
+	SNATTarget       = "SNAT"
 
 	PreRoutingChain  = "PREROUTING"
 	ForwardChain     = "FORWARD"
@@ -49,6 +50,14 @@ const (
 	waitIntervalMicroSeconds = 200000
 )
 
+type Protocol byte
+
+const (
+	ProtocolDual Protocol = iota
+	ProtocolIPv4
+	ProtocolIPv6
+)
+
 // https://netfilter.org/projects/iptables/files/changes-iptables-1.6.2.txt:
 // iptables-restore: support acquiring the lock.
 var restoreWaitSupportedMinVersion = semver.Version{Major: 1, Minor: 6, Patch: 2}
@@ -141,6 +150,40 @@ func (c *Client) EnsureRule(table string, chain string, ruleSpec []string) error
 	return nil
 }
 
+// InsertRule checks if target rule already exists, inserts it if not.
+func (c *Client) InsertRule(protocol Protocol, table string, chain string, ruleSpec []string) error {
+	for idx := range c.ipts {
+		ipt := c.ipts[idx]
+		if !matchProtocol(ipt, protocol) {
+			continue
+		}
+		exist, err := ipt.Exists(table, chain, ruleSpec...)
+		if err != nil {
+			return fmt.Errorf("error checking if rule %v exists in table %s chain %s: %v", ruleSpec, table, chain, err)
+		}
+		if exist {
+			return nil
+		}
+		if err := ipt.Insert(table, chain, 1, ruleSpec...); err != nil {
+			return fmt.Errorf("error inserting rule %v to table %s chain %s: %v", ruleSpec, table, chain, err)
+		}
+	}
+	klog.V(2).Infof("Inserted rule %v to table %s chain %s", ruleSpec, table, chain)
+	return nil
+}
+
+func matchProtocol(ipt *iptables.IPTables, protocol Protocol) bool {
+	switch protocol {
+	case ProtocolDual:
+		return true
+	case ProtocolIPv4:
+		return ipt.Proto() == iptables.ProtocolIPv4
+	case ProtocolIPv6:
+		return ipt.Proto() == iptables.ProtocolIPv6
+	}
+	return false
+}
+
 // DeleteRule checks if target rule already exists, deletes the rule if found.
 func (c *Client) DeleteRule(table string, chain string, ruleSpec []string) error {
 	for idx := range c.ipts {

test/integration/agent/route_test.go

@@ -252,11 +252,17 @@ func TestIpTablesSync(t *testing.T) {
 	select {
 	case <-inited: // Node network initialized
 	}
+
+	snatIP := net.ParseIP("1.1.1.1")
+	mark := uint32(1)
+	assert.NoError(t, routeClient.AddSNATRule(snatIP, mark))
+
 	tcs := []struct {
 		RuleSpec, Cmd, Table, Chain string
 	}{
 		{Table: "raw", Cmd: "-A", Chain: "OUTPUT", RuleSpec: "-m comment --comment \"Antrea: jump to Antrea output rules\" -j ANTREA-OUTPUT"},
 		{Table: "filter", Cmd: "-A", Chain: "ANTREA-FORWARD", RuleSpec: "-i antrea-gw0 -m comment --comment \"Antrea: accept packets from local pods\" -j ACCEPT"},
+		{Table: "nat", Cmd: "-A", Chain: "ANTREA-POSTROUTING", RuleSpec: fmt.Sprintf("-m comment --comment \"Antrea: snat pod to external packets\" -m mark --mark %#x/0xff -j SNAT --to-source %s", mark, snatIP)},
 	}
 	// we delete some rules, start the sync goroutine, wait for sync operation to restore them.
 	for _, tc := range tcs {
@@ -281,6 +287,41 @@ func TestIpTablesSync(t *testing.T) {
 	close(stopCh)
 }
 
+func TestAddAndDeleteSNATRule(t *testing.T) {
+	skipIfNotInContainer(t)
+	gwLink := createDummyGW(t)
+	defer netlink.LinkDel(gwLink)
+
+	routeClient, err := route.NewClient(serviceCIDR, &config.NetworkConfig{TrafficEncapMode: config.TrafficEncapModeEncap}, false)
+	assert.Nil(t, err)
+
+	inited := make(chan struct{})
+	err = routeClient.Initialize(nodeConfig, func() {
+		close(inited)
+	})
+	assert.NoError(t, err)
+	select {
+	case <-inited: // Node network initialized
+	}
+
+	snatIP := net.ParseIP("1.1.1.1")
+	mark := uint32(1)
+	expectedRule := fmt.Sprintf("-m comment --comment \"Antrea: snat pod to external packets\" -m mark --mark %#x/0xff -j SNAT --to-source %s", mark, snatIP)
+
+	assert.NoError(t, routeClient.AddSNATRule(snatIP, mark))
+	saveCmd := fmt.Sprintf("iptables-save -t nat | grep ANTREA-POSTROUTING")
+	// #nosec G204: ignore in test code
+	actualData, err := exec.Command("bash", "-c", saveCmd).Output()
+	assert.NoError(t, err, "error executing iptables-save cmd")
+	assert.Contains(t, string(actualData), expectedRule)
+
+	assert.NoError(t, routeClient.DeleteSNATRule(mark))
+	// #nosec G204: ignore in test code
+	actualData, err = exec.Command("bash", "-c", saveCmd).Output()
+	assert.NoError(t, err, "error executing iptables-save cmd")
+	assert.NotContains(t, string(actualData), expectedRule)
+}
+
 func TestAddAndDeleteRoutes(t *testing.T) {
 	skipIfNotInContainer(t)