Merge pull request #35 from coroot/adaptive_connection_filtering

Added heuristics to track connections with public destination IPs

common/net.go

@@ -1,9 +1,29 @@
 package common
 
 import (
+	"github.com/coroot/coroot-node-agent/flags"
 	"inet.af/netaddr"
+	"k8s.io/klog/v2"
 )
 
+var (
+	ConnectionFilter = connectionFilter{
+		whitelist: map[string]netaddr.IPPrefix{},
+	}
+)
+
+func init() {
+	if flags.ExternalNetworksWhitelist != nil {
+		for _, prefix := range *flags.ExternalNetworksWhitelist {
+			p, err := netaddr.ParseIPPrefix(prefix)
+			if err != nil {
+				klog.Fatalf("invalid network %s: %s", prefix, err)
+			}
+			ConnectionFilter.WhitelistPrefix(p)
+		}
+	}
+}
+
 func IsIpPrivate(ip netaddr.IP) bool {
 	if ip.IsPrivate() {
 		return true
@@ -14,3 +34,44 @@ func IsIpPrivate(ip netaddr.IP) bool {
 	}
 	return false
 }
+
+type connectionFilter struct {
+	whitelist map[string]netaddr.IPPrefix
+}
+
+func (f connectionFilter) WhitelistIP(ip netaddr.IP) {
+	var bits uint8 = 32
+	if ip.Is6() {
+		bits = 128
+	}
+	f.WhitelistPrefix(netaddr.IPPrefixFrom(ip, bits))
+}
+
+func (f connectionFilter) WhitelistPrefix(p netaddr.IPPrefix) {
+	if _, ok := f.whitelist[p.String()]; ok {
+		return
+	}
+	f.whitelist[p.String()] = p
+}
+
+func (f connectionFilter) ShouldBeSkipped(dst, actualDst netaddr.IP) bool {
+	if IsIpPrivate(dst) || dst.IsLoopback() {
+		return false
+	}
+	for _, prefix := range f.whitelist {
+		if prefix.Contains(dst) {
+			return false
+		}
+	}
+	if IsIpPrivate(actualDst) || actualDst.IsLoopback() {
+		f.WhitelistIP(dst)
+		return false
+	}
+	for _, prefix := range f.whitelist {
+		if prefix.Contains(actualDst) {
+			f.WhitelistIP(dst)
+			return false
+		}
+	}
+	return true
+}

common/net_test.go

@@ -0,0 +1,27 @@
+package common
+
+import (
+	"github.com/stretchr/testify/assert"
+	"inet.af/netaddr"
+	"testing"
+)
+
+func TestConnectionFilter(t *testing.T) {
+	f := connectionFilter{whitelist: map[string]netaddr.IPPrefix{}}
+	assert.False(t, f.ShouldBeSkipped(netaddr.MustParseIP("127.0.0.1"), netaddr.MustParseIP("127.0.0.1")))
+	assert.False(t, f.ShouldBeSkipped(netaddr.MustParseIP("192.168.1.1"), netaddr.MustParseIP("127.0.0.1")))
+
+	assert.True(t, f.ShouldBeSkipped(netaddr.MustParseIP("1.1.1.1"), netaddr.MustParseIP("2.2.2.2")))
+	assert.False(t, f.ShouldBeSkipped(netaddr.MustParseIP("1.1.1.1"), netaddr.MustParseIP("192.168.1.1")))
+	// because the actual dest is allowed, the dest is added to whitelist
+	assert.False(t, f.ShouldBeSkipped(netaddr.MustParseIP("1.1.1.1"), netaddr.MustParseIP("2.2.2.2")))
+
+	assert.True(t, f.ShouldBeSkipped(netaddr.MustParseIP("2.2.2.2"), netaddr.MustParseIP("2.2.2.2")))
+	f.WhitelistPrefix(netaddr.MustParseIPPrefix("2.2.2.0/24"))
+	assert.False(t, f.ShouldBeSkipped(netaddr.MustParseIP("2.2.2.2"), netaddr.MustParseIP("2.2.2.2")))
+
+	assert.True(t, f.ShouldBeSkipped(netaddr.MustParseIP("3.3.3.3"), netaddr.MustParseIP("3.3.3.3")))
+	f.WhitelistPrefix(netaddr.MustParseIPPrefix("4.4.4.4/32"))
+	assert.False(t, f.ShouldBeSkipped(netaddr.MustParseIP("3.3.3.3"), netaddr.MustParseIP("4.4.4.4")))
+
+}

containers/container.go

@@ -487,14 +487,20 @@ func (c *Container) onConnectionOpen(pid uint32, fd uint64, src, dst netaddr.IPP
 	if dst.IP().IsLoopback() && !p.isHostNs() {
 		return
 	}
-	whitelisted := false
-	for _, prefix := range flags.ExternalNetworksWhitelist {
-		if prefix.Contains(dst.IP()) {
-			whitelisted = true
-			break
+	actualDst, err := c.getActualDestination(p, src, dst)
+	if err != nil {
+		if !common.IsNotExist(err) {
+			klog.Warningf("cannot open NetNs for pid %d: %s", pid, err)
 		}
+		return
 	}
-	if !whitelisted && !common.IsIpPrivate(dst.IP()) && !dst.IP().IsLoopback() {
+	switch {
+	case actualDst == nil:
+		actualDst = &dst
+	case actualDst.IP().IsLoopback() && !p.isHostNs():
+		return
+	}
+	if common.ConnectionFilter.ShouldBeSkipped(dst.IP(), actualDst.IP()) {
 		return
 	}
 	c.lock.Lock()
@@ -502,19 +508,6 @@ func (c *Container) onConnectionOpen(pid uint32, fd uint64, src, dst netaddr.IPP
 	if failed {
 		c.connectsFailed[dst]++
 	} else {
-		actualDst, err := c.getActualDestination(p, src, dst)
-		if err != nil {
-			if !common.IsNotExist(err) {
-				klog.Warningf("cannot open NetNs for pid %d: %s", pid, err)
-			}
-			return
-		}
-		switch {
-		case actualDst == nil:
-			actualDst = &dst
-		case actualDst.IP().IsLoopback() && !p.isHostNs():
-			return
-		}
 		c.connectsSuccessful[AddrPair{src: dst, dst: *actualDst}]++
 		c.connectionsActive[AddrPair{src: src, dst: dst}] = &ActiveConnection{
 			ActualDest:         *actualDst,

flags/flags.go

@@ -2,8 +2,6 @@ package flags
 
 import (
 	"gopkg.in/alecthomas/kingpin.v2"
-	"inet.af/netaddr"
-	"k8s.io/klog/v2"
 	"os"
 	"strings"
 )
@@ -15,8 +13,7 @@ var (
 	DisablePinger     = kingpin.Flag("disable-pinger", "Don't ping upstreams").Default("false").Bool()
 	DisableL7Tracing  = kingpin.Flag("disable-l7-tracing", "Disable L7 tracing").Default("false").Bool()
 
-	externalNetworksWhitelist = kingpin.Flag("track-public-network", "Allow track connections to the specified IP networks, all private networks are allowed by default (e.g., Y.Y.Y.Y/mask)").Strings()
-	ExternalNetworksWhitelist []netaddr.IPPrefix
+	ExternalNetworksWhitelist = kingpin.Flag("track-public-network", "Allow track connections to the specified IP networks, all private networks are allowed by default (e.g., Y.Y.Y.Y/mask)").Strings()
 
 	Provider          = kingpin.Flag("provider", "`provider` label for `node_cloud_info` metric").Envar("PROVIDER").String()
 	Region            = kingpin.Flag("region", "`region` label for `node_cloud_info` metric").Envar("REGION").String()
@@ -38,13 +35,4 @@ func init() {
 	}
 	kingpin.HelpFlag.Short('h').Hidden()
 	kingpin.Parse()
-	if externalNetworksWhitelist != nil {
-		for _, prefix := range *externalNetworksWhitelist {
-			p, err := netaddr.ParseIPPrefix(prefix)
-			if err != nil {
-				klog.Fatalf("invalid network %s: %s", prefix, err)
-			}
-			ExternalNetworksWhitelist = append(ExternalNetworksWhitelist, p)
-		}
-	}
 }

main.go

@@ -80,15 +80,13 @@ func whitelistNodeExternalNetworks() {
 		klog.Warningln("failed to get network interfaces:", err)
 		return
 	}
-	seenPrefixes := map[string]bool{}
 	for _, iface := range netdevs {
 		for _, p := range iface.IPPrefixes {
-			if p.IP().IsLoopback() || common.IsIpPrivate(p.IP()) || seenPrefixes[p.String()] {
+			if p.IP().IsLoopback() || common.IsIpPrivate(p.IP()) {
 				continue
 			}
 			// if the node has an external network IP, whitelist that network
-			flags.ExternalNetworksWhitelist = append(flags.ExternalNetworksWhitelist, p)
-			seenPrefixes[p.String()] = true
+			common.ConnectionFilter.WhitelistPrefix(p)
 		}
 	}
 }