Delete stale UDP conntrack entries when adding new Portmaps to containers

go.mod

@@ -23,11 +23,10 @@ require (
 	github.com/safchain/ethtool v0.0.0-20190326074333-42ed695e3de8
 	github.com/sirupsen/logrus v1.0.6 // indirect
 	github.com/stretchr/testify v1.3.0 // indirect
-	github.com/vishvananda/netlink v1.1.0
-	github.com/vishvananda/netns v0.0.0-20200728191858-db3c7e526aae // indirect
+	github.com/vishvananda/netlink v1.1.1-0.20201029203352-d40f9887b852
 	golang.org/x/crypto v0.0.0-20181009213950-7c1a557ab941 // indirect
 	golang.org/x/net v0.0.0-20181011144130-49bb7cea24b1 // indirect
-	golang.org/x/sys v0.0.0-20200831180312-196b9ba8737a
+	golang.org/x/sys v0.0.0-20201117170446-d9b008d0a637
 	gopkg.in/airbrake/gobrake.v2 v2.0.9 // indirect
 	gopkg.in/gemnasium/logrus-airbrake-hook.v2 v2.1.2 // indirect
 )

go.sum

@@ -43,26 +43,18 @@ github.com/sirupsen/logrus v1.0.6/go.mod h1:pMByvHTf9Beacp5x1UXfOR9xyW/9antXMhjM
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/testify v1.3.0 h1:TivCn/peBQ7UY8ooIcPgZFpTNSz0Q2U6UrFlUfqbe0Q=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
-github.com/vishvananda/netlink v0.0.0-20181108222139-023a6dafdcdf h1:3J37+NPjNyGW/dbfXtj3yWuF9OEepIdGOXRaJGbORV8=
-github.com/vishvananda/netlink v0.0.0-20181108222139-023a6dafdcdf/go.mod h1:+SR5DhBJrl6ZM7CoCKvpw5BKroDKQ+PJqOg65H/2ktk=
-github.com/vishvananda/netlink v1.1.0 h1:1iyaYNBLmP6L0220aDnYQpo1QEV4t4hJ+xEEhhJH8j0=
-github.com/vishvananda/netlink v1.1.0/go.mod h1:cTgwzPIzzgDAYoQrMm0EdrjRUBkTqKYppBueQtXaqoE=
-github.com/vishvananda/netns v0.0.0-20180720170159-13995c7128cc h1:R83G5ikgLMxrBvLh22JhdfI8K6YXEPHx5P03Uu3DRs4=
-github.com/vishvananda/netns v0.0.0-20180720170159-13995c7128cc/go.mod h1:ZjcWmFBXmLKZu9Nxj3WKYEafiSqer2rnvPr0en9UNpI=
-github.com/vishvananda/netns v0.0.0-20191106174202-0a2b9b5464df h1:OviZH7qLw/7ZovXvuNyL3XQl8UFofeikI1NW1Gypu7k=
-github.com/vishvananda/netns v0.0.0-20191106174202-0a2b9b5464df/go.mod h1:JP3t17pCcGlemwknint6hfoeCVQrEMVwxRLRjXpq+BU=
+github.com/vishvananda/netlink v1.1.1-0.20201029203352-d40f9887b852 h1:cPXZWzzG0NllBLdjWoD1nDfaqu98YMv+OneaKc8sPOA=
+github.com/vishvananda/netlink v1.1.1-0.20201029203352-d40f9887b852/go.mod h1:twkDnbuQxJYemMlGd4JFIcuhgX83tXhKS2B/PRMpOho=
 github.com/vishvananda/netns v0.0.0-20200728191858-db3c7e526aae h1:4hwBBUfQCFe3Cym0ZtKyq7L16eZUtYKs+BaHDN6mAns=
 github.com/vishvananda/netns v0.0.0-20200728191858-db3c7e526aae/go.mod h1:DD4vA1DwXk04H54A1oHXtwZmA0grkVMdPxx/VGLCah0=
 golang.org/x/crypto v0.0.0-20181009213950-7c1a557ab941 h1:qBTHLajHecfu+xzRI9PqVDcqx7SdHj9d4B+EzSn3tAc=
 golang.org/x/crypto v0.0.0-20181009213950-7c1a557ab941/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/net v0.0.0-20181011144130-49bb7cea24b1 h1:Y/KGZSOdz/2r0WJ9Mkmz6NJBusp0kiNx1Cn82lzJQ6w=
 golang.org/x/net v0.0.0-20181011144130-49bb7cea24b1/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
-golang.org/x/sys v0.0.0-20190606203320-7fc4e5ec1444/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20190616124812-15dcb6c0061f h1:25KHgbfyiSm6vwQLbM3zZIe1v9p/3ea4Rz+nnM5K/i4=
-golang.org/x/sys v0.0.0-20190616124812-15dcb6c0061f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200217220822-9197077df867/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20200831180312-196b9ba8737a h1:i47hUS795cOydZI4AwJQCKXOr4BvxzvikwDoDtHhP2Y=
-golang.org/x/sys v0.0.0-20200831180312-196b9ba8737a/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20200728102440-3e129f6d46b1/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20201117170446-d9b008d0a637 h1:O5hKNaGxIT4A8OTMnuh6UpmBdI3SAPxlZ3g0olDrJVM=
+golang.org/x/sys v0.0.0-20201117170446-d9b008d0a637/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 gopkg.in/airbrake/gobrake.v2 v2.0.9 h1:7z2uVWwn7oVeeugY1DtlPAy5H+KYgB1KeKTnqjNatLo=
 gopkg.in/airbrake/gobrake.v2 v2.0.9/go.mod h1:/h5ZAUhDkGaJfjzjKLSjv6zCL6O0LLBxU4K+aSYdM/U=
 gopkg.in/gemnasium/logrus-airbrake-hook.v2 v2.1.2 h1:OAj3g0cR6Dx/R07QgQe8wkA9RNjB2u4i700xBkIT4e0=

pkg/testutils/echo/client/client.go

@@ -10,28 +10,40 @@ import (
 func main() {
 	target := flag.String("target", "", "the server address")
 	payload := flag.String("message", "", "the message to send to the server")
+	protocol := flag.String("protocol", "tcp", "the protocol to use with the server [udp,tcp], default tcp")
 	flag.Parse()
 
 	if *target == "" || *payload == "" {
 		flag.Usage()
 		panic("invalid arguments")
 	}
-	conn, err := net.Dial("tcp", *target)
+
+	switch *protocol {
+	case "tcp":
+		connectTCP(*target, *payload)
+	case "udp":
+		connectUDP(*target, *payload)
+	default:
+		panic("invalid protocol")
+	}
+}
+
+func connectTCP(target, payload string) {
+	conn, err := net.Dial("tcp", target)
 	if err != nil {
-		panic(fmt.Sprintf("Failed to open connection to [%s] %v", *target, err))
+		panic(fmt.Sprintf("Failed to open connection to [%s] %v", target, err))
 	}
 	defer conn.Close()
 
-	_, err = conn.Write([]byte(*payload))
+	_, err = conn.Write([]byte(payload))
 	if err != nil {
 		panic("Failed to send payload")
 	}
 	_, err = conn.Write([]byte("\n"))
 	if err != nil {
 		panic("Failed to send payload")
 	}
-
-	buf := make([]byte, 4)
+	buf := make([]byte, 1024)
 	for {
 		n, err := conn.Read(buf)
 		fmt.Print(string(buf[:n]))
@@ -43,3 +55,36 @@ func main() {
 		}
 	}
 }
+
+// UDP uses a constant source port to trigger conntrack problems
+func connectUDP(target, payload string) {
+	LocalAddr, err := net.ResolveUDPAddr("udp", ":54321")
+	if err != nil {
+		panic(fmt.Sprintf("Failed to resolve UDP local address on port 54321 %v", err))
+	}
+	RemoteAddr, err := net.ResolveUDPAddr("udp", target)
+	if err != nil {
+		panic(fmt.Sprintf("Failed to resolve UDP remote address [%s] %v", target, err))
+	}
+	conn, err := net.DialUDP("udp", LocalAddr, RemoteAddr)
+	if err != nil {
+		panic(fmt.Sprintf("Failed to open connection to [%s] %v", target, err))
+	}
+	defer conn.Close()
+
+	_, err = conn.Write([]byte(payload))
+	if err != nil {
+		panic("Failed to send payload")
+	}
+	_, err = conn.Write([]byte("\n"))
+	if err != nil {
+		panic("Failed to send payload")
+	}
+
+	buf := make([]byte, 1024)
+	n, err := conn.Read(buf)
+	if err != nil {
+		panic("Failed to read from socket")
+	}
+	fmt.Print(string(buf[:n]))
+}

pkg/testutils/echo/server/main.go

@@ -10,28 +10,58 @@ import (
 	"bufio"
 	"fmt"
 	"io"
+	"log"
 	"net"
 	"os"
 	"strings"
 	"time"
 )
 
 func main() {
+	// Start TCP server
 	listener, err := net.Listen("tcp", ":")
 	if err != nil {
 		panic(err)
 	}
+	defer listener.Close()
+	// use the same port for UDP
 	_, port, err := net.SplitHostPort(listener.Addr().String())
 	if err != nil {
 		panic(err)
 	}
 	fmt.Printf("127.0.0.1:%s\n", port)
+	go func() {
+		for {
+			conn, err := listener.Accept()
+			if err != nil {
+				panic(err)
+			}
+			go handleConnection(conn)
+		}
+	}()
+
+	// Start UDP server
+	addr, err := net.ResolveUDPAddr("udp", fmt.Sprintf(":%s", port))
+	if err != nil {
+		log.Fatalf("Error from net.ResolveUDPAddr(): %s", err)
+	}
+	sock, err := net.ListenUDP("udp", addr)
+	if err != nil {
+		log.Fatalf("Error from ListenUDP(): %s", err)
+	}
+	defer sock.Close()
+
+	buffer := make([]byte, 1024)
 	for {
-		conn, err := listener.Accept()
+		n, addr, err := sock.ReadFrom(buffer)
 		if err != nil {
-			panic(err)
+			log.Fatalf("Error from ReadFrom(): %s", err)
+		}
+		sock.SetWriteDeadline(time.Now().Add(1 * time.Minute))
+		n, err = sock.WriteTo(buffer[0:n], addr)
+		if err != nil {
+			return
 		}
-		go handleConnection(conn)
 	}
 }
 
@@ -53,5 +83,4 @@ func handleConnection(conn net.Conn) {
 		fmt.Fprint(os.Stderr, err.Error())
 		return
 	}
-
 }

pkg/utils/conntrack.go

@@ -0,0 +1,73 @@
+// Copyright 2020 CNI authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package utils
+
+import (
+	"fmt"
+	"net"
+
+	"github.com/vishvananda/netlink"
+	"golang.org/x/sys/unix"
+)
+
+// Assigned Internet Protocol Numbers
+// https://www.iana.org/assignments/protocol-numbers/protocol-numbers.xhtml
+const (
+	PROTOCOL_TCP  = 6
+	PROTOCOL_UDP  = 17
+	PROTOCOL_SCTP = 132
+)
+
+// getNetlinkFamily returns the Netlink IP family constant
+func getNetlinkFamily(isIPv6 bool) netlink.InetFamily {
+	if isIPv6 {
+		return unix.AF_INET6
+	}
+	return unix.AF_INET
+}
+
+// DeleteConntrackEntriesForDstIP delete the conntrack entries for the connections
+// specified by the given destination IP and protocol
+func DeleteConntrackEntriesForDstIP(dstIP string, protocol uint8) error {
+	ip := net.ParseIP(dstIP)
+	if ip == nil {
+		return fmt.Errorf("error deleting connection tracking state, bad IP %s", ip)
+	}
+	family := getNetlinkFamily(ip.To4() == nil)
+
+	filter := &netlink.ConntrackFilter{}
+	filter.AddIP(netlink.ConntrackOrigDstIP, ip)
+	filter.AddProtocol(protocol)
+
+	_, err := netlink.ConntrackDeleteFilter(netlink.ConntrackTable, family, filter)
+	if err != nil {
+		return fmt.Errorf("error deleting connection tracking state for protocol: %d IP: %s, error: %v", protocol, ip, err)
+	}
+	return nil
+}
+
+// DeleteConntrackEntriesForDstPort delete the conntrack entries for the connections specified
+// by the given destination port, protocol and IP family
+func DeleteConntrackEntriesForDstPort(port uint16, protocol uint8, family netlink.InetFamily) error {
+	filter := &netlink.ConntrackFilter{}
+	filter.AddPort(netlink.ConntrackOrigDstPort, port)
+	filter.AddProtocol(protocol)
+
+	_, err := netlink.ConntrackDeleteFilter(netlink.ConntrackTable, family, filter)
+	if err != nil {
+		return fmt.Errorf("error deleting connection tracking state for protocol: %d Port: %d, error: %v", protocol, port, err)
+	}
+	return nil
+}

plugins/meta/portmap/main.go

@@ -28,12 +28,14 @@ package main
 import (
 	"encoding/json"
 	"fmt"
+	"log"
 	"net"
 
 	"github.com/containernetworking/cni/pkg/skel"
 	"github.com/containernetworking/cni/pkg/types"
 	"github.com/containernetworking/cni/pkg/types/current"
 	"github.com/containernetworking/cni/pkg/version"
+	"golang.org/x/sys/unix"
 
 	bv "github.com/containernetworking/plugins/pkg/utils/buildversion"
 )
@@ -89,12 +91,24 @@ func cmdAdd(args *skel.CmdArgs) error {
 		if err := forwardPorts(netConf, netConf.ContIPv4); err != nil {
 			return err
 		}
+		// Delete conntrack entries for UDP to avoid conntrack blackholing traffic
+		// due to stale connections. We do that after the iptables rules are set, so
+		// the new traffic uses them. Failures are informative only.
+		if err := deletePortmapStaleConnections(netConf.RuntimeConfig.PortMaps, unix.AF_INET); err != nil {
+			log.Printf("failed to delete stale UDP conntrack entries for %s: %v", netConf.ContIPv4.IP, err)
+		}
 	}
 
 	if netConf.ContIPv6.IP != nil {
 		if err := forwardPorts(netConf, netConf.ContIPv6); err != nil {
 			return err
 		}
+		// Delete conntrack entries for UDP to avoid conntrack blackholing traffic
+		// due to stale connections. We do that after the iptables rules are set, so
+		// the new traffic uses them. Failures are informative only.
+		if err := deletePortmapStaleConnections(netConf.RuntimeConfig.PortMaps, unix.AF_INET6); err != nil {
+			log.Printf("failed to delete stale UDP conntrack entries for %s: %v", netConf.ContIPv6.IP, err)
+		}
 	}
 
 	// Pass through the previous result

plugins/meta/portmap/portmap.go

@@ -19,10 +19,12 @@ import (
 	"net"
 	"sort"
 	"strconv"
+	"strings"
 
 	"github.com/containernetworking/plugins/pkg/utils"
 	"github.com/containernetworking/plugins/pkg/utils/sysctl"
 	"github.com/coreos/go-iptables/iptables"
+	"github.com/vishvananda/netlink"
 )
 
 // This creates the chains to be added to iptables. The basic structure is
@@ -42,10 +44,12 @@ import (
 // The names of the top-level summary chains.
 // These should never be changed, or else upgrading will require manual
 // intervention.
-const TopLevelDNATChainName = "CNI-HOSTPORT-DNAT"
-const SetMarkChainName = "CNI-HOSTPORT-SETMARK"
-const MarkMasqChainName = "CNI-HOSTPORT-MASQ"
-const OldTopLevelSNATChainName = "CNI-HOSTPORT-SNAT"
+const (
+	TopLevelDNATChainName    = "CNI-HOSTPORT-DNAT"
+	SetMarkChainName         = "CNI-HOSTPORT-SETMARK"
+	MarkMasqChainName        = "CNI-HOSTPORT-MASQ"
+	OldTopLevelSNATChainName = "CNI-HOSTPORT-SNAT"
+)
 
 // forwardPorts establishes port forwarding to a given container IP.
 // containerNet.IP can be either v4 or v6.
@@ -113,7 +117,6 @@ func forwardPorts(config *PortMapConf, containerNet net.IPNet) error {
 }
 
 func checkPorts(config *PortMapConf, containerNet net.IPNet) error {
-
 	dnatChain := genDnatChain(config.Name, config.ContainerID)
 	fillDnatRules(&dnatChain, config, containerNet)
 
@@ -189,7 +192,7 @@ func fillDnatRules(c *chain, config *PortMapConf, containerNet net.IPNet) {
 		setMarkChainName = *config.ExternalSetMarkChain
 	}
 
-	//Generate the dnat entry rules. We'll use multiport, but it ony accepts
+	// Generate the dnat entry rules. We'll use multiport, but it ony accepts
 	// up to 15 rules, so partition the list if needed.
 	// Do it in a stable order for testing
 	protoPorts := groupByProto(entries)
@@ -243,7 +246,8 @@ func fillDnatRules(c *chain, config *PortMapConf, containerNet net.IPNet) {
 
 		ruleBase := []string{
 			"-p", entry.Protocol,
-			"--dport", strconv.Itoa(entry.HostPort)}
+			"--dport", strconv.Itoa(entry.HostPort),
+		}
 		if addRuleBaseDst {
 			ruleBase = append(ruleBase,
 				"-d", entry.HostIP)
@@ -406,3 +410,19 @@ func maybeGetIptables(isV6 bool) *iptables.IPTables {
 
 	return ipt
 }
+
+// deletePortmapStaleConnections delete the UDP conntrack entries on the specified IP family
+// from the ports mapped to the container
+func deletePortmapStaleConnections(portMappings []PortMapEntry, family netlink.InetFamily) error {
+	for _, pm := range portMappings {
+		// skip if is not UDP
+		if strings.ToLower(pm.Protocol) != "udp" {
+			continue
+		}
+		err := utils.DeleteConntrackEntriesForDstPort(uint16(pm.HostPort), utils.PROTOCOL_UDP, family)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}