enable network policies strict mode

apply network policies to existing connections.

The dataplane now inspect the existing connections in the conntrack
table and evaluates against the current network policies.
If a established connection is no longer allowed then the dataplane
sets the conntrack entry timeout to zero, causing the subsequent packets
to be enqueued and processed dropping them if are no longer enabled.

The strict mode is enabled by default and runs at most every 30 seconds
once there is a change triggered in the dataplane, this is to avoid
performance issues for listing conntrack entries too often.

Author: aojea

cmd/kube-network-policies/iptracker/main.go

@@ -123,6 +123,7 @@ func run() int {
 		FailOpen:            opts.FailOpen,
 		QueueID:             opts.QueueID,
 		NetfilterBug1766Fix: opts.NetfilterBug1766Fix,
+		StrictMode:          opts.StrictMode,
 	}
 
 	var config *rest.Config

cmd/kube-network-policies/npa-v1alpha2/main.go

@@ -83,6 +83,7 @@ func run() int {
 		FailOpen:            opts.FailOpen,
 		QueueID:             opts.QueueID,
 		NetfilterBug1766Fix: opts.NetfilterBug1766Fix,
+		StrictMode:          opts.StrictMode,
 	}
 
 	var config *rest.Config

cmd/kube-network-policies/standard/main.go

@@ -78,6 +78,7 @@ func run() int {
 		FailOpen:            opts.FailOpen,
 		QueueID:             opts.QueueID,
 		NetfilterBug1766Fix: opts.NetfilterBug1766Fix,
+		StrictMode:          opts.StrictMode,
 	}
 
 	var config *rest.Config

pkg/cmd/cmd.go

@@ -25,6 +25,7 @@ type Options struct {
 	HostnameOverride    string
 	NetfilterBug1766Fix bool
 	DisableNRI          bool
+	StrictMode          bool
 }
 
 // NewOptions creates a new Options object with default values.
@@ -41,6 +42,7 @@ func (o *Options) AddFlags(fs *flag.FlagSet) {
 	fs.StringVar(&o.HostnameOverride, "hostname-override", "", "If non-empty, will be used as the name of the Node that kube-network-policies is running on. If unset, the node name is assumed to be the same as the node's hostname.")
 	fs.BoolVar(&o.NetfilterBug1766Fix, "netfilter-bug-1766-fix", true, "If set, process DNS packets on the PREROUTING hooks to avoid the race condition on the conntrack subsystem, not needed for kernels 6.12+ (see https://bugzilla.netfilter.org/show_bug.cgi?id=1766)")
 	fs.BoolVar(&o.DisableNRI, "disable-nri", false, "If set, disable NRI, that is used to get the Pod IP information directly from the runtime to avoid the race explained in https://issues.k8s.io/85966")
+	fs.BoolVar(&o.StrictMode, "strict-mode", true, "If set, changes to network policies also affect established connections")
 
 	fs.Usage = func() {
 		fmt.Fprint(os.Stderr, "Usage: kube-network-policies [options]\n\n")

pkg/dataplane/conntrack.go

@@ -0,0 +1,49 @@
+package dataplane
+
+import (
+	"github.com/vishvananda/netlink"
+	"golang.org/x/sys/unix"
+	v1 "k8s.io/api/core/v1"
+	"k8s.io/klog/v2"
+	"sigs.k8s.io/kube-network-policies/pkg/network"
+)
+
+var (
+	mapIPFamilyToString = map[uint8]v1.IPFamily{
+		unix.AF_INET:  v1.IPv4Protocol,
+		unix.AF_INET6: v1.IPv6Protocol,
+	}
+	mapProtocolToString = map[uint8]v1.Protocol{
+		unix.IPPROTO_TCP:  v1.ProtocolTCP,
+		unix.IPPROTO_UDP:  v1.ProtocolUDP,
+		unix.IPPROTO_SCTP: v1.ProtocolSCTP,
+	}
+)
+
+func PacketFromFlow(flow *netlink.ConntrackFlow) *network.Packet {
+	if flow == nil {
+		return nil
+	}
+	packet := network.Packet{
+		SrcIP:   flow.Forward.SrcIP,
+		DstIP:   flow.Reverse.SrcIP,
+		SrcPort: int(flow.Forward.SrcPort),
+		DstPort: int(flow.Reverse.SrcPort),
+	}
+
+	if family, ok := mapIPFamilyToString[flow.FamilyType]; ok {
+		packet.Family = family
+	} else {
+		klog.InfoS("Unknown IP family", "family", flow.FamilyType, "flow", flow)
+		return nil
+	}
+
+	if protocol, ok := mapProtocolToString[flow.Forward.Protocol]; ok {
+		packet.Proto = protocol
+	} else {
+		klog.InfoS("Unknown protocol", "protocol", flow.Forward.Protocol, "flow", flow)
+		return nil
+	}
+
+	return &packet
+}

pkg/dataplane/controller.go

@@ -2,6 +2,7 @@ package dataplane
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"time"
 
@@ -10,9 +11,12 @@ import (
 	"github.com/google/nftables/binaryutil"
 	"github.com/google/nftables/expr"
 	"github.com/mdlayher/netlink"
+	vishnetlink "github.com/vishvananda/netlink"
+	"github.com/vishvananda/netlink/nl"
 	"golang.org/x/sys/unix"
 
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/klog/v2"
 
@@ -49,6 +53,7 @@ type Config struct {
 	QueueID             int
 	NetfilterBug1766Fix bool
 	NFTableName         string // if other projects use this controllers they need to be able to use their own table name
+	StrictMode          bool   // enforce network policies also on established connections
 }
 
 func (c *Config) Defaults() error {
@@ -99,9 +104,25 @@ func newController(
 		1*time.Hour,   // maxInterval
 	)
 
+	if c.config.StrictMode {
+		// The runner will explore existing connections listed on the conntrack table
+		// and timeout the conntrack entries of the no longer valid connections to reenqueue
+		// the packets and enforce network policies.
+		c.connRunner = runner.NewBoundedFrequencyRunner(
+			controllerName+"-firewall-enforcer",
+			func() error { return c.firewallEnforcer(context.Background()) },
+			30*time.Second, // minInterval (less frequent than nftables sync to avoid overload listing conntrack entries)
+			15*time.Second, // retryInterval
+			1*time.Hour,    // maxInterval
+		)
+	}
+
 	// The sync callback now triggers the runner.
 	syncCallback := func() {
 		c.syncRunner.Run()
+		if c.config.StrictMode {
+			c.connRunner.Run()
+		}
 	}
 	c.policyEngine.SetDataplaneSyncCallbacks(syncCallback)
 
@@ -113,6 +134,7 @@ type Controller struct {
 	config       Config
 	policyEngine *networkpolicy.PolicyEngine
 	syncRunner   *runner.BoundedFrequencyRunner
+	connRunner   *runner.BoundedFrequencyRunner
 
 	nfq     *nfqueue.Nfqueue
 	flushed bool
@@ -142,7 +164,7 @@ func (c *Controller) Run(ctx context.Context) error {
 	registerMetrics(ctx)
 	// collect metrics periodically
 	go wait.UntilWithContext(ctx, func(ctx context.Context) {
-		logger := klog.FromContext(ctx)
+		logger := klog.FromContext(ctx).WithName("metrics-collector")
 		queues, err := readNfnetlinkQueueStats()
 		if err != nil {
 			logger.Error(err, "reading nfqueue stats")
@@ -159,8 +181,11 @@ func (c *Controller) Run(ctx context.Context) error {
 
 	}, 30*time.Second)
 
-	// Start the BoundedFrequencyRunner's loop.
+	// Start the BoundedFrequencyRunner's loops.
 	go c.syncRunner.Loop(ctx.Done())
+	if c.config.StrictMode {
+		go c.connRunner.Loop(ctx.Done())
+	}
 
 	// Perform an initial sync to ensure rules are in place at startup.
 	if err := c.syncNFTablesRules(ctx); err != nil {
@@ -310,11 +335,99 @@ func (c *Controller) evaluatePacket(ctx context.Context, p *network.Packet) bool
 	return allowed
 }
 
+// firewallEnforcer retrieves conntrack entries and enforces current network policies on them
+// by flushing the conntrack entries that are not allowed anymore so they are
+// processed again in the queue.
+func (c *Controller) firewallEnforcer(ctx context.Context) error {
+	var errorList []error
+	logger := klog.FromContext(ctx).WithName("firewall-enforcer")
+	logger.Info("Enforcing firewall policies on existing connections")
+
+	start := time.Now()
+
+	flows, err := vishnetlink.ConntrackTableList(vishnetlink.ConntrackTable, vishnetlink.FAMILY_ALL)
+	if err != nil {
+		logger.Error(err, "listing conntrack entries")
+		return err
+	}
+
+	defer func() {
+		logger.Info("Completed enforcing firewall policies on existing connections", "nflows", len(flows), "elapsed", time.Since(start))
+	}()
+
+	allPodIPs, divertAll, err := c.policyEngine.GetManagedIPs(ctx)
+	if err != nil {
+		logger.Error(err, "getting managed IPs for firewall enforcement")
+		return err
+	}
+
+	ipset := sets.Set[string]{}
+	if !divertAll {
+		for _, ip := range allPodIPs {
+			ipset.Insert(ip.String())
+		}
+	}
+
+	for _, flow := range flows {
+		// only UDP, SCTP or TCP connections in ESTABLISHED state are evaluated
+		if flow.Forward.Protocol != unix.IPPROTO_UDP &&
+			flow.Forward.Protocol != unix.IPPROTO_SCTP &&
+			flow.Forward.Protocol != unix.IPPROTO_TCP {
+			continue
+		}
+		if flow.ProtoInfo != nil {
+			if state, ok := flow.ProtoInfo.(*vishnetlink.ProtoInfoTCP); ok && state.State != nl.TCP_CONNTRACK_ESTABLISHED {
+				continue
+			}
+		}
+
+		// If divertAll is true, all pod IPs are managed by network policies.
+		if !divertAll {
+			// Only evaluate flows that are affected by network policies.
+			// It checks the source IP of the forward flow and the translated IP of the reverse flow,
+			// as these are the IPs that belong to the pods in case of DNAT for Services.
+			if !ipset.Has(flow.Forward.SrcIP.String()) && !ipset.Has(flow.Reverse.SrcIP.String()) {
+				logger.V(4).Info("Skipping conntrack entry not involving managed IPs", "flow", flow)
+				continue
+			}
+		}
+
+		// The policy engine evaluates packets, so we need to convert the conntrack flow to a packet.
+		// The packet is evaluated against the current network policies both for source and destination.
+		packet := PacketFromFlow(flow)
+		if packet == nil {
+			continue
+		}
+		logger.V(4).Info("Evaluating packet", "packet", packet.String())
+
+		// Evaluate the packet against current network policies.
+		allowed, err := c.policyEngine.EvaluatePacket(ctx, packet)
+		if err != nil {
+			logger.Info("error evaluating conntrack entry", "flow", flow, "err", err)
+			continue
+		}
+		// If the flow is not allowed, timeout the connection so it is removed immediately.
+		// Flushing the connection will cause the packets to be re-queued and evaluated again.
+		// Deleting the conntrack entry will not work if tcp_loose is enabled in the kernel
+		// since the connection would be re-established immediately by the kernel.
+		if !allowed {
+			logger.V(4).Info("Connection no longer allowed by network policies", "packet", packet.String())
+			flow.TimeOut = 0
+			err = vishnetlink.ConntrackUpdate(vishnetlink.ConntrackTable, vishnetlink.InetFamily(flow.FamilyType), flow)
+			if err != nil {
+				errorList = append(errorList, err)
+			}
+		}
+	}
+
+	return errors.Join(errorList...)
+}
+
 // syncNFTablesRules adds the necessary rules to process the first connection packets in userspace
 // and check if network policies must apply.
 // TODO: We can divert only the traffic affected by network policies using a set in nftables or an IPset.
 func (c *Controller) syncNFTablesRules(ctx context.Context) error {
-	logger := klog.FromContext(ctx)
+	logger := klog.FromContext(ctx).WithName("nftables-sync")
 
 	logger.Info("Syncing nftables rules")
 	start := time.Now()
@@ -580,7 +693,7 @@ func (c *Controller) syncNFTablesRules(ctx context.Context) error {
 	}
 
 	if err := nft.Flush(); err != nil {
-		klog.FromContext(ctx).Info("syncing nftables rules", "error", err)
+		logger.Info("syncing nftables rules", "error", err)
 		return err
 	}
 	return nil

tests/README.md

@@ -5,4 +5,24 @@
 
 2. Install `kind` https://kind.sigs.k8s.io/
 
-3. Run `bats tests/`
+3. Run `bats tests/`
+
+## Troubleshooting test failures
+
+`bats -x -o _artifacts --print-output-on-failure --filter "network policy drops established connections" tests/e2e_standard.bats`
+
+You can modify or comment the `tests/setup_suite.bash` hooks to avoid creating and recreating the cluster.
+
+```diff
+diff --git a/tests/setup_suite.bash b/tests/setup_suite.bash
+index f34cc39..8006903 100644
+--- a/tests/setup_suite.bash
++++ b/tests/setup_suite.bash
+@@ -29,5 +29,5 @@ EOF
+
+ function teardown_suite {
+     kind export logs "$BATS_TEST_DIRNAME"/../_artifacts --name "$CLUSTER_NAME"
+-    kind delete cluster --name "$CLUSTER_NAME"
++    # kind delete cluster --name "$CLUSTER_NAME"
+ }
+ ```