Flush container flows in conntrack (Bug #8795)

Flush all the endpoint flows when the external
connectivity is removed.
This will prevent issues where if there is a flow
in conntrack this will have precedence and will
let the packet skip the POSTROUTING chain.

Signed-off-by: Flavio Crisciani <flavio.crisciani@docker.com>

Author: Flavio Crisciani

drivers/bridge/bridge.go

@@ -1346,6 +1346,13 @@ func (d *driver) RevokeExternalConnectivity(nid, eid string) error {
 
 	endpoint.portMapping = nil
 
+	// Clean the connection tracker state of the host for the specific endpoint
+	// The host kernel keeps track of the connections (TCP and UDP), so if a new endpoint gets the same IP of
+	// this one (that is going down), is possible that some of the packets would not be routed correctly inside
+	// the new endpoint
+	// Deeper details: https://github.com/docker/docker/issues/8795
+	clearEndpointConnections(d.nlh, endpoint)
+
 	if err = d.storeUpdate(endpoint); err != nil {
 		return fmt.Errorf("failed to update bridge endpoint %s to store: %v", endpoint.id[0:7], err)
 	}

drivers/bridge/setup_ip_tables.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/Sirupsen/logrus"
 	"github.com/docker/libnetwork/iptables"
+	"github.com/vishvananda/netlink"
 )
 
 // DockerChain: DOCKER iptable chain name
@@ -348,3 +349,15 @@ func setupInternalNetworkRules(bridgeIface string, addr net.Addr, icc, insert bo
 	}
 	return nil
 }
+
+func clearEndpointConnections(nlh *netlink.Handle, ep *bridgeEndpoint) {
+	var ipv4List []net.IP
+	var ipv6List []net.IP
+	if ep.addr != nil {
+		ipv4List = append(ipv4List, ep.addr.IP)
+	}
+	if ep.addrv6 != nil {
+		ipv6List = append(ipv6List, ep.addrv6.IP)
+	}
+	iptables.DeleteConntrackEntries(nlh, ipv4List, ipv6List)
+}

iptables/conntrack.go

@@ -0,0 +1,59 @@
+package iptables
+
+import (
+	"errors"
+	"net"
+	"syscall"
+
+	"github.com/Sirupsen/logrus"
+	"github.com/vishvananda/netlink"
+)
+
+var (
+	// ErrConntrackNotConfigurable means that conntrack module is not loaded or does not have the netlink module loaded
+	ErrConntrackNotConfigurable = errors.New("conntrack is not available")
+)
+
+// IsConntrackProgrammable returns true if the handle supports the NETLINK_NETFILTER and the base modules are loaded
+func IsConntrackProgrammable(nlh *netlink.Handle) bool {
+	return nlh.SupportsNetlinkFamily(syscall.NETLINK_NETFILTER)
+}
+
+// DeleteConntrackEntries deletes all the conntrack connections on the host for the specified IP
+// Returns the number of flows deleted for IPv4, IPv6 else error
+func DeleteConntrackEntries(nlh *netlink.Handle, ipv4List []net.IP, ipv6List []net.IP) (uint, uint, error) {
+	if !IsConntrackProgrammable(nlh) {
+		return 0, 0, ErrConntrackNotConfigurable
+	}
+
+	var totalIPv4FlowPurged uint
+	for _, ipAddress := range ipv4List {
+		flowPurged, err := purgeConntrackState(nlh, syscall.AF_INET, ipAddress)
+		if err != nil {
+			logrus.Warnf("Failed to delete conntrack state for %s: %v", ipAddress, err)
+			continue
+		}
+		totalIPv4FlowPurged += flowPurged
+	}
+
+	var totalIPv6FlowPurged uint
+	for _, ipAddress := range ipv6List {
+		flowPurged, err := purgeConntrackState(nlh, syscall.AF_INET6, ipAddress)
+		if err != nil {
+			logrus.Warnf("Failed to delete conntrack state for %s: %v", ipAddress, err)
+			continue
+		}
+		totalIPv6FlowPurged += flowPurged
+	}
+
+	logrus.Debugf("DeleteConntrackEntries purged ipv4:%d, ipv6:%d", totalIPv4FlowPurged, totalIPv6FlowPurged)
+	return totalIPv4FlowPurged, totalIPv6FlowPurged, nil
+}
+
+func purgeConntrackState(nlh *netlink.Handle, family netlink.InetFamily, ipAddress net.IP) (uint, error) {
+	filter := &netlink.ConntrackFilter{}
+	// NOTE: doing the flush using the ipAddress is safe because today there cannot be multiple networks with the same subnet
+	// so it will not be possible to flush flows that are of other containers
+	filter.AddIP(netlink.ConntrackNatAnyIP, ipAddress)
+	return nlh.ConntrackDeleteFilter(netlink.ConntrackTable, family, filter)
+}

iptables/iptables.go

@@ -100,14 +100,14 @@ func detectIptables() {
 	supportsCOpt = supportsCOption(mj, mn, mc)
 }
 
-func initIptables() {
+func initDependencies() {
 	probe()
 	initFirewalld()
 	detectIptables()
 }
 
 func initCheck() error {
-	initOnce.Do(initIptables)
+	initOnce.Do(initDependencies)
 
 	if iptablesPath == "" {
 		return ErrIptablesNotFound

ns/init_linux.go

@@ -75,13 +75,28 @@ func NlHandle() *netlink.Handle {
 
 func getSupportedNlFamilies() []int {
 	fams := []int{syscall.NETLINK_ROUTE}
+	// NETLINK_XFRM test
 	if err := loadXfrmModules(); err != nil {
 		if checkXfrmSocket() != nil {
 			logrus.Warnf("Could not load necessary modules for IPSEC rules: %v", err)
-			return fams
+		} else {
+			fams = append(fams, syscall.NETLINK_XFRM)
 		}
+	} else {
+		fams = append(fams, syscall.NETLINK_XFRM)
 	}
-	return append(fams, syscall.NETLINK_XFRM)
+	// NETLINK_NETFILTER test
+	if err := loadNfConntrackModules(); err != nil {
+		if checkNfSocket() != nil {
+			logrus.Warnf("Could not load necessary modules for Conntrack: %v", err)
+		} else {
+			fams = append(fams, syscall.NETLINK_NETFILTER)
+		}
+	} else {
+		fams = append(fams, syscall.NETLINK_NETFILTER)
+	}
+
+	return fams
 }
 
 func loadXfrmModules() error {
@@ -103,3 +118,23 @@ func checkXfrmSocket() error {
 	syscall.Close(fd)
 	return nil
 }
+
+func loadNfConntrackModules() error {
+	if out, err := exec.Command("modprobe", "-va", "nf_conntrack").CombinedOutput(); err != nil {
+		return fmt.Errorf("Running modprobe nf_conntrack failed with message: `%s`, error: %v", strings.TrimSpace(string(out)), err)
+	}
+	if out, err := exec.Command("modprobe", "-va", "nf_conntrack_netlink").CombinedOutput(); err != nil {
+		return fmt.Errorf("Running modprobe nf_conntrack_netlink failed with message: `%s`, error: %v", strings.TrimSpace(string(out)), err)
+	}
+	return nil
+}
+
+// API check on required nf_conntrack* modules (nf_conntrack, nf_conntrack_netlink)
+func checkNfSocket() error {
+	fd, err := syscall.Socket(syscall.AF_NETLINK, syscall.SOCK_RAW, syscall.NETLINK_NETFILTER)
+	if err != nil {
+		return err
+	}
+	syscall.Close(fd)
+	return nil
+}