firewall: Keep conntrack entries per peer and recheck them when a peer reconnects

firewall.go

@@ -32,10 +32,11 @@ type conn struct {
 	Expires time.Time // Time when this conntrack entry will expire
 
 	// record why the original connection passed the firewall, so we can re-validate
-	// after ruleset changes. Note, rulesVersion is a uint16 so that these two
+	// after ruleset changes or when the peer handshakes again, which may present a
-	// fields pack for free after the uint32 above
+	// cert with different groups. These pack into the word after Expires.
 	incoming     bool
 	rulesVersion uint16
+	localIndex   uint32
 }
 
 // TODO: need conntrack max tracked connections handling
@@ -82,8 +83,8 @@ type firewallMetrics struct {
 type FirewallConntrack struct {
 	sync.Mutex
 
-	Conns      map[firewall.Packet]*conn
+	Conns      map[firewall.ConntrackKey]*conn
-	TimerWheel *TimerWheel[firewall.Packet]
+	TimerWheel *TimerWheel[firewall.ConntrackKey]
 }
 
 // FirewallTable is the entry point for a rule, the evaluation order is:
@@ -167,8 +168,8 @@ func NewFirewall(l *slog.Logger, tcpTimeout, UDPTimeout, defaultTimeout time.Dur
 
 	return &Firewall{
 		Conntrack: &FirewallConntrack{
-			Conns:      make(map[firewall.Packet]*conn),
+			Conns:      make(map[firewall.ConntrackKey]*conn),
-			TimerWheel: NewTimerWheel[firewall.Packet](tmin, tmax),
+			TimerWheel: NewTimerWheel[firewall.ConntrackKey](tmin, tmax),
 		},
 		InRules:          newFirewallTable(),
 		OutRules:         newFirewallTable(),
@@ -457,8 +458,12 @@ func (f *Firewall) Drop(fp firewall.Packet, incoming bool, h *HostInfo, caPool *
 		return ErrInvalidLocalIP
 	}
 
+	// The highest vpn address identifies the peer. v2 certs sort their networks and v1 certs have only one,
+	// so this is the peer's v6 address when it has one, which stays put while v4 addresses come and go.
+	key := firewall.ConntrackKey{Packet: fp, Peer: h.vpnAddrs[len(h.vpnAddrs)-1]}
+
 	// Check if we spoke to this tuple, if we did then allow this packet
-	if f.inConns(fp, h, caPool, localCache) {
+	if f.inConns(key, h, caPool, localCache) {
 		return nil
 	}
 
@@ -474,7 +479,7 @@ func (f *Firewall) Drop(fp firewall.Packet, incoming bool, h *HostInfo, caPool *
 	}
 
 	// We always want to conntrack since it is a faster operation
-	f.addConn(fp, incoming)
+	f.addConn(key, incoming, h)
 
 	return nil
 }
@@ -503,9 +508,9 @@ func (f *Firewall) EmitStats() {
 	metrics.GetOrRegisterGauge("firewall.rules.hash", nil).Update(int64(f.GetRuleHashFNV()))
 }
 
-func (f *Firewall) inConns(fp firewall.Packet, h *HostInfo, caPool *cert.CAPool, localCache firewall.ConntrackCache) bool {
+func (f *Firewall) inConns(key firewall.ConntrackKey, h *HostInfo, caPool *cert.CAPool, localCache firewall.ConntrackCache) bool {
 	if localCache != nil {
-		if _, ok := localCache[fp]; ok {
+		if _, ok := localCache[key]; ok {
 			return true
 		}
 	}
@@ -518,49 +523,52 @@ func (f *Firewall) inConns(fp firewall.Packet, h *HostInfo, caPool *cert.CAPool,
 		f.evict(ep)
 	}
 
-	c, ok := conntrack.Conns[fp]
+	c, ok := conntrack.Conns[key]
 
 	if !ok {
 		conntrack.Unlock()
 		return false
 	}
 
-	if c.rulesVersion != f.rulesVersion {
+	if c.rulesVersion != f.rulesVersion || c.localIndex != h.localIndexId {
-		// This conntrack entry was for an older rule set, validate
+		// This conntrack entry was allowed by an older rule set or for an earlier tunnel to the peer, validate
-		// it still passes with the current rule set
+		// it still passes with the current rule set and the peer's current certificate
 		table := f.OutRules
 		if c.incoming {
 			table = f.InRules
 		}
 
 		// We now know which firewall table to check against
-		if !table.match(fp, c.incoming, h.ConnectionState.peerCert, caPool) {
+		if !table.match(key.Packet, c.incoming, h.ConnectionState.peerCert, caPool) {
 			if f.l.Enabled(context.Background(), slog.LevelDebug) {
-				h.logger(f.l).Debug("dropping old conntrack entry, does not match new ruleset",
+				h.logger(f.l).Debug("dropping old conntrack entry, does not match current ruleset and certificate",
-					"fwPacket", fp,
+					"fwPacket", key.Packet,
 					"incoming", c.incoming,
 					"rulesVersion", f.rulesVersion,
 					"oldRulesVersion", c.rulesVersion,
+					"oldLocalIndex", c.localIndex,
 				)
 			}
-			delete(conntrack.Conns, fp)
+			delete(conntrack.Conns, key)
 			conntrack.Unlock()
 			return false
 		}
 
 		if f.l.Enabled(context.Background(), slog.LevelDebug) {
-			h.logger(f.l).Debug("keeping old conntrack entry, does match new ruleset",
+			h.logger(f.l).Debug("keeping old conntrack entry, does match current ruleset and certificate",
-				"fwPacket", fp,
+				"fwPacket", key.Packet,
 				"incoming", c.incoming,
 				"rulesVersion", f.rulesVersion,
 				"oldRulesVersion", c.rulesVersion,
+				"oldLocalIndex", c.localIndex,
 			)
 		}
 
 		c.rulesVersion = f.rulesVersion
+		c.localIndex = h.localIndexId
 	}
 
-	switch fp.Protocol {
+	switch key.Protocol {
 	case iputil.IPProtocolTCP:
 		c.Expires = time.Now().Add(f.TCPTimeout)
 	case iputil.IPProtocolUDP:
@@ -572,17 +580,17 @@ func (f *Firewall) inConns(fp firewall.Packet, h *HostInfo, caPool *cert.CAPool,
 	conntrack.Unlock()
 
 	if localCache != nil {
-		localCache[fp] = struct{}{}
+		localCache[key] = struct{}{}
 	}
 
 	return true
 }
 
-func (f *Firewall) addConn(fp firewall.Packet, incoming bool) {
+func (f *Firewall) addConn(key firewall.ConntrackKey, incoming bool, h *HostInfo) {
 	var timeout time.Duration
 	c := &conn{}
 
-	switch fp.Protocol {
+	switch key.Protocol {
 	case iputil.IPProtocolTCP:
 		timeout = f.TCPTimeout
 	case iputil.IPProtocolUDP:
@@ -593,26 +601,27 @@ func (f *Firewall) addConn(fp firewall.Packet, incoming bool) {
 
 	conntrack := f.Conntrack
 	conntrack.Lock()
-	if _, ok := conntrack.Conns[fp]; !ok {
+	if _, ok := conntrack.Conns[key]; !ok {
 		conntrack.TimerWheel.Advance(time.Now())
-		conntrack.TimerWheel.Add(fp, timeout)
+		conntrack.TimerWheel.Add(key, timeout)
 	}
 
-	// Record which rulesVersion allowed this connection, so we can retest after
+	// Record which rulesVersion and tunnel allowed this connection, so we can retest after
-	// firewall reload
+	// firewall reload or when the peer handshakes again
 	c.incoming = incoming
 	c.rulesVersion = f.rulesVersion
+	c.localIndex = h.localIndexId
 	c.Expires = time.Now().Add(timeout)
-	conntrack.Conns[fp] = c
+	conntrack.Conns[key] = c
 	conntrack.Unlock()
 }
 
 // Evict checks if a conntrack entry has expired, if so it is removed, if not it is re-added to the wheel
 // Caller must own the connMutex lock!
-func (f *Firewall) evict(p firewall.Packet) {
+func (f *Firewall) evict(key firewall.ConntrackKey) {
 	// Are we still tracking this conn?
 	conntrack := f.Conntrack
-	t, ok := conntrack.Conns[p]
+	t, ok := conntrack.Conns[key]
 	if !ok {
 		return
 	}
@@ -622,12 +631,12 @@ func (f *Firewall) evict(p firewall.Packet) {
 	// Timeout is in the future, re-add the timer
 	if newT > 0 {
 		conntrack.TimerWheel.Advance(time.Now())
-		conntrack.TimerWheel.Add(p, newT)
+		conntrack.TimerWheel.Add(key, newT)
 		return
 	}
 
 	// This conn is done
-	delete(conntrack.Conns, p)
+	delete(conntrack.Conns, key)
 }
 
 func (ft *FirewallTable) match(p firewall.Packet, incoming bool, c *cert.CachedCertificate, caPool *cert.CAPool) bool {

firewall/cache.go

@@ -3,15 +3,23 @@ package firewall
 import (
 	"context"
 	"log/slog"
+	"net/netip"
 	"sync/atomic"
 	"time"
 
 	"github.com/slackhq/nebula/logging"
 )
 
+// ConntrackKey identifies a tracked flow by its packet and the peer it was allowed for, so peers that route
+// the same unsafe network never share a conntrack entry.
+type ConntrackKey struct {
+	Peer netip.Addr
+	Packet
+}
+
 // ConntrackCache is used as a local routine cache to know if a given flow
 // has been seen in the conntrack table.
-type ConntrackCache map[Packet]struct{}
+type ConntrackCache map[ConntrackKey]struct{}
 
 type ConntrackCacheTicker struct {
 	cacheV    uint64

firewall/cache_test.go

@@ -23,7 +23,7 @@ func newFixedTicker(t *testing.T, l *slog.Logger, cacheLen int) *ConntrackCacheT
 		cache: make(ConntrackCache, cacheLen),
 	}
 	for i := range cacheLen {
-		c.cache[Packet{LocalPort: uint16(i) + 1}] = struct{}{}
+		c.cache[ConntrackKey{Packet: Packet{LocalPort: uint16(i) + 1}}] = struct{}{}
 	}
 	c.cacheTick.Store(1) // cacheV starts at 0, so Get() takes the reset path
 	return c

firewall/packet.go

@@ -26,6 +26,10 @@ type Packet struct {
 	RemotePort uint16
 	Protocol   uint8
 	Fragment   bool
+
+	// pad fills what would otherwise be trailing padding so maps hash and compare a ConntrackKey as one
+	// block of memory instead of field by field. It must always be zero.
+	pad [2]byte
 }
 
 func (fp *Packet) Copy() *Packet {

firewall_test.go

@@ -3,6 +3,7 @@ package nebula
 import (
 	"bytes"
 	"errors"
+	"fmt"
 	"log/slog"
 	"math"
 	"net/netip"
@@ -979,6 +980,149 @@ func TestFirewall_ConntrackSourceSpoofingAcrossPeers(t *testing.T) {
 		"SECURITY: attacker spoofed victim's overlay source IP (192.0.2.2) by reusing an existing conntrack 4-tuple; Drop returned %v instead of rejecting", got)
 }
 
+func TestFirewall_ConntrackUnsafeNetworkPeers(t *testing.T) {
+	l := test.NewLoggerWithOutput(&bytes.Buffer{})
+
+	myVpnNetworksTable := new(bart.Lite)
+	myVpnNetworksTable.Insert(netip.MustParsePrefix("192.0.2.1/24"))
+
+	owner := &dummyCert{
+		name:     "owner",
+		networks: []netip.Prefix{netip.MustParsePrefix("192.0.2.1/24")},
+	}
+
+	// Two routers for the same unsafe network, only one of them is trusted inbound
+	router := func(name string, addr string, groups ...string) *HostInfo {
+		inverted := map[string]struct{}{}
+		for _, g := range groups {
+			inverted[g] = struct{}{}
+		}
+		c := &cert.CachedCertificate{
+			Certificate: &dummyCert{
+				name:           name,
+				groups:         groups,
+				networks:       []netip.Prefix{netip.MustParsePrefix(addr + "/24")},
+				unsafeNetworks: []netip.Prefix{netip.MustParsePrefix("198.51.100.0/24")},
+			},
+			InvertedGroups: inverted,
+		}
+		h := &HostInfo{
+			ConnectionState: &ConnectionState{peerCert: c},
+			vpnAddrs:        []netip.Addr{netip.MustParseAddr(addr)},
+		}
+		h.buildNetworks(myVpnNetworksTable, c.Certificate)
+		return h
+	}
+	trusted := router("trusted", "192.0.2.2", "trusted")
+	other := router("other", "192.0.2.3")
+
+	cp := cert.NewCAPool()
+	flow := firewall.Packet{
+		LocalAddr:  netip.MustParseAddr("192.0.2.1"),
+		RemoteAddr: netip.MustParseAddr("198.51.100.5"),
+		LocalPort:  443,
+		RemotePort: 55000,
+		Protocol:   iputil.IPProtocolTCP,
+	}
+
+	for _, useCache := range []bool{false, true} {
+		var cache firewall.ConntrackCache
+		if useCache {
+			cache = firewall.ConntrackCache{}
+		}
+
+		t.Run(fmt.Sprintf("inbound flow allowed for one peer/cache=%v", useCache), func(t *testing.T) {
+			fw := NewFirewall(l, time.Second, time.Minute, time.Hour, owner)
+			require.NoError(t, fw.AddRule(true, firewall.ProtoAny, 0, 0, []string{"trusted"}, "", "", "", "", ""))
+			clear(cache)
+
+			require.NoError(t, fw.Drop(flow, true, trusted, cp, cache))
+			assert.Equal(t, ErrNoMatchingRule, fw.Drop(flow, true, other, cp, cache))
+			require.NoError(t, fw.Drop(flow, true, trusted, cp, cache))
+		})
+
+		t.Run(fmt.Sprintf("reply to a flow opened via one peer/cache=%v", useCache), func(t *testing.T) {
+			fw := NewFirewall(l, time.Second, time.Minute, time.Hour, owner)
+			require.NoError(t, fw.AddRule(false, firewall.ProtoAny, 0, 0, []string{"any"}, "", "", "", "", ""))
+			clear(cache)
+
+			require.NoError(t, fw.Drop(flow, false, trusted, cp, cache))
+			require.NoError(t, fw.Drop(flow, true, trusted, cp, cache))
+			assert.Equal(t, ErrNoMatchingRule, fw.Drop(flow, true, other, cp, cache))
+		})
+	}
+}
+
+func TestFirewall_ConntrackPeerRehandshake(t *testing.T) {
+	l := test.NewLoggerWithOutput(&bytes.Buffer{})
+
+	myVpnNetworksTable := new(bart.Lite)
+	myVpnNetworksTable.Insert(netip.MustParsePrefix("192.0.2.1/24"))
+
+	owner := &dummyCert{
+		name:     "owner",
+		networks: []netip.Prefix{netip.MustParsePrefix("192.0.2.1/24")},
+	}
+
+	// Each call is a new tunnel to the same peer, presenting a cert with the given groups
+	tunnel := func(localIndex uint32, groups ...string) *HostInfo {
+		inverted := map[string]struct{}{}
+		for _, g := range groups {
+			inverted[g] = struct{}{}
+		}
+		c := &cert.CachedCertificate{
+			Certificate: &dummyCert{
+				name:     "peer",
+				groups:   groups,
+				networks: []netip.Prefix{netip.MustParsePrefix("192.0.2.2/24"), netip.MustParsePrefix("2001:db8::2/64")},
+			},
+			InvertedGroups: inverted,
+		}
+		h := &HostInfo{
+			ConnectionState: &ConnectionState{peerCert: c},
+			vpnAddrs:        []netip.Addr{netip.MustParseAddr("192.0.2.2"), netip.MustParseAddr("2001:db8::2")},
+			localIndexId:    localIndex,
+		}
+		h.buildNetworks(myVpnNetworksTable, c.Certificate)
+		return h
+	}
+
+	cp := cert.NewCAPool()
+	flow := firewall.Packet{
+		LocalAddr:  netip.MustParseAddr("192.0.2.1"),
+		RemoteAddr: netip.MustParseAddr("192.0.2.2"),
+		LocalPort:  443,
+		RemotePort: 55000,
+		Protocol:   iputil.IPProtocolTCP,
+	}
+	key := firewall.ConntrackKey{Packet: flow, Peer: netip.MustParseAddr("2001:db8::2")}
+
+	t.Run("inbound flow follows the peer's groups", func(t *testing.T) {
+		fw := NewFirewall(l, time.Second, time.Minute, time.Hour, owner)
+		require.NoError(t, fw.AddRule(true, firewall.ProtoAny, 0, 0, []string{"trusted"}, "", "", "", "", ""))
+
+		require.NoError(t, fw.Drop(flow, true, tunnel(1, "trusted"), cp, nil))
+
+		// Same groups on a new tunnel keeps the entry and moves it to the new tunnel
+		require.NoError(t, fw.Drop(flow, true, tunnel(2, "trusted"), cp, nil))
+		require.Contains(t, fw.Conntrack.Conns, key)
+		assert.Equal(t, uint32(2), fw.Conntrack.Conns[key].localIndex)
+
+		// Losing the group on a new tunnel removes the entry
+		assert.Equal(t, ErrNoMatchingRule, fw.Drop(flow, true, tunnel(3), cp, nil))
+		assert.NotContains(t, fw.Conntrack.Conns, key)
+	})
+
+	t.Run("reply to a flow we opened survives a new tunnel", func(t *testing.T) {
+		fw := NewFirewall(l, time.Second, time.Minute, time.Hour, owner)
+		require.NoError(t, fw.AddRule(false, firewall.ProtoAny, 0, 0, []string{"any"}, "", "", "", "", ""))
+
+		require.NoError(t, fw.Drop(flow, false, tunnel(1, "trusted"), cp, nil))
+		require.NoError(t, fw.Drop(flow, true, tunnel(2), cp, nil))
+		assert.Equal(t, uint32(2), fw.Conntrack.Conns[key].localIndex)
+	})
+}
+
 // BenchmarkFirewallDropConntrackHit measures Drop on an already-established flow
 // (a conntrack hit). This is the fast path that the source-IP<->cert binding
 // reordering adds work to, so it quantifies the cost of moving the address checks
@@ -1742,6 +1886,6 @@ func (mf *mockFirewall) AddRule(incoming bool, proto uint8, startPort int32, end
 
 func resetConntrack(fw *Firewall) {
 	fw.Conntrack.Lock()
-	fw.Conntrack.Conns = map[firewall.Packet]*conn{}
+	fw.Conntrack.Conns = map[firewall.ConntrackKey]*conn{}
 	fw.Conntrack.Unlock()
 }

interface_emit_test.go

@@ -39,7 +39,7 @@ func Test_statsEmitter_setsGauges(t *testing.T) {
 		hostMap:          hostMap,
 		inside:           &overlaytest.NoopTun{},
 		outside:          &udp.NoopConn{},
-		firewall:         &Firewall{Conntrack: &FirewallConntrack{Conns: map[firewall.Packet]*conn{}}},
+		firewall:         &Firewall{Conntrack: &FirewallConntrack{Conns: map[firewall.ConntrackKey]*conn{}}},
 		lightHouse:       lh,
 		pki:              &PKI{},
 		handshakeManager: NewHandshakeManager(l, hostMap, lh, &udp.NoopConn{}, defaultHandshakeConfig),