diff --git a/firewall.go b/firewall.go index 5fb2d6de..3a3d31a2 100644 --- a/firewall.go +++ b/firewall.go @@ -14,6 +14,7 @@ import ( "strconv" "strings" "sync" + "sync/atomic" "time" "github.com/gaissmai/bart" @@ -28,16 +29,23 @@ type FirewallInterface interface { AddRule(incoming bool, proto uint8, startPort int32, endPort int32, groups []string, host string, cidr, localCidr string, caName string, caSha string) error } +// conn is a conntrack entry. Only expires changes after it is stored, a rules version change +// stores a replacement entry instead. type conn struct { - Expires time.Time // Time when this conntrack entry will expire + expires atomic.Uint32 // conntrack clock second when this entry expires, see FirewallConntrack.now // 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 - // fields pack for free after the uint32 above + // after ruleset changes incoming bool rulesVersion uint16 } +func newConn(expires uint32, incoming bool, rulesVersion uint16) *conn { + c := &conn{incoming: incoming, rulesVersion: rulesVersion} + c.expires.Store(expires) + return c +} + // TODO: need conntrack max tracked connections handling type Firewall struct { Conntrack *FirewallConntrack @@ -79,11 +87,69 @@ type firewallMetrics struct { droppedNoRule metrics.Counter } +// FirewallConntrack is safe for concurrent use by every routine without a lock. type FirewallConntrack struct { - sync.Mutex + conns sync.Map // firewall.Packet -> *conn + count atomic.Int64 // entries in conns, which sync.Map can't tell us + now atomic.Uint32 // the conntrack clock: whole seconds since epoch, kept current by Run + epoch time.Time +} - Conns map[firewall.Packet]*conn - TimerWheel *TimerWheel[firewall.Packet] +func newFirewallConntrack() *FirewallConntrack { + return &FirewallConntrack{epoch: time.Now()} +} + +// Run advances the conntrack clock every second and sweeps out expired entries every minute, +// until ctx is done. Reading a shared clock is much cheaper than asking the OS on every packet. +func (ct *FirewallConntrack) Run(ctx context.Context) { + t := time.NewTicker(time.Second) + defer t.Stop() + var nextSweep uint32 + for { + select { + case <-ctx.Done(): + return + case <-t.C: + // Measured from epoch rather than counted, so a late tick can't slow the clock down + now := uint32(time.Since(ct.epoch) / time.Second) + ct.now.Store(now) + if now >= nextSweep { + ct.sweep(now) + nextSweep = now + 60 + } + } + } +} + +// sweep deletes entries that expired at or before now. +func (ct *FirewallConntrack) sweep(now uint32) { + ct.conns.Range(func(k, v any) bool { + if v.(*conn).expires.Load() <= now { + ct.delete(k.(firewall.Packet), v.(*conn)) + } + return true + }) +} + +func (ct *FirewallConntrack) store(fp firewall.Packet, c *conn) { + if _, replaced := ct.conns.Swap(fp, c); !replaced { + ct.count.Add(1) + } +} + +// delete removes fp's entry if it is still c, so a newer entry stored by another routine survives +func (ct *FirewallConntrack) delete(fp firewall.Packet, c *conn) { + if ct.conns.CompareAndDelete(fp, c) { + ct.count.Add(-1) + } +} + +// clear deletes every entry +func (ct *FirewallConntrack) clear() { + ct.conns.Range(func(k, v any) bool { + ct.delete(k.(firewall.Packet), v.(*conn)) + return true + }) } // FirewallTable is the entry point for a rule, the evaluation order is: @@ -132,26 +198,10 @@ type firewallLocalCIDR struct { LocalCIDR *bart.Lite } -// NewFirewall creates a new Firewall object. A TimerWheel is created for you from the provided timeouts. +// NewFirewall creates a new Firewall object. // The certificate provided should be the highest version loaded in memory. func NewFirewall(l *slog.Logger, tcpTimeout, UDPTimeout, defaultTimeout time.Duration, c cert.Certificate) *Firewall { //TODO: error on 0 duration - var tmin, tmax time.Duration - - if tcpTimeout < UDPTimeout { - tmin = tcpTimeout - tmax = UDPTimeout - } else { - tmin = UDPTimeout - tmax = tcpTimeout - } - - if defaultTimeout < tmin { - tmin = defaultTimeout - } else if defaultTimeout > tmax { - tmax = defaultTimeout - } - routableNetworks := new(bart.Lite) var assignedNetworks []netip.Prefix for _, network := range c.Networks() { @@ -166,10 +216,7 @@ func NewFirewall(l *slog.Logger, tcpTimeout, UDPTimeout, defaultTimeout time.Dur } return &Firewall{ - Conntrack: &FirewallConntrack{ - Conns: make(map[firewall.Packet]*conn), - TimerWheel: NewTimerWheel[firewall.Packet](tmin, tmax), - }, + Conntrack: newFirewallConntrack(), InRules: newFirewallTable(), OutRules: newFirewallTable(), TCPTimeout: tcpTimeout, @@ -423,7 +470,7 @@ var ErrNoMatchingRule = errors.New("no matching rule in firewall table") // Drop returns an error if the packet should be dropped, explaining why. It // returns nil if the packet should not be dropped. -func (f *Firewall) Drop(fp firewall.Packet, incoming bool, h *HostInfo, caPool *cert.CAPool, localCache firewall.ConntrackCache) error { +func (f *Firewall) Drop(fp firewall.Packet, incoming bool, h *HostInfo, caPool *cert.CAPool) error { // Make sure remote address matches nebula certificate, and determine how to treat it if h.networks == nil { // Simple case: Certificate has one address and no unsafe networks @@ -458,7 +505,7 @@ func (f *Firewall) Drop(fp firewall.Packet, incoming bool, h *HostInfo, caPool * } // Check if we spoke to this tuple, if we did then allow this packet - if f.inConns(fp, h, caPool, localCache) { + if f.inConns(fp, h, caPool) { return nil } @@ -494,36 +541,19 @@ func (f *Firewall) Destroy() { } func (f *Firewall) EmitStats() { - conntrack := f.Conntrack - conntrack.Lock() - conntrackCount := len(conntrack.Conns) - conntrack.Unlock() - metrics.GetOrRegisterGauge("firewall.conntrack.count", nil).Update(int64(conntrackCount)) + metrics.GetOrRegisterGauge("firewall.conntrack.count", nil).Update(f.Conntrack.count.Load()) metrics.GetOrRegisterGauge("firewall.rules.version", nil).Update(int64(f.rulesVersion)) 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 { - if localCache != nil { - if _, ok := localCache[fp]; ok { - return true - } - } +func (f *Firewall) inConns(fp firewall.Packet, h *HostInfo, caPool *cert.CAPool) bool { conntrack := f.Conntrack - conntrack.Lock() - - // Purge every time we test - ep, has := conntrack.TimerWheel.Purge() - if has { - f.evict(ep) - } - - c, ok := conntrack.Conns[fp] - + v, ok := conntrack.conns.Load(fp) if !ok { - conntrack.Unlock() return false } + c := v.(*conn) + expires := conntrack.now.Load() + f.timeout(fp.Protocol) if c.rulesVersion != f.rulesVersion { // This conntrack entry was for an older rule set, validate @@ -543,8 +573,7 @@ func (f *Firewall) inConns(fp firewall.Packet, h *HostInfo, caPool *cert.CAPool, "oldRulesVersion", c.rulesVersion, ) } - delete(conntrack.Conns, fp) - conntrack.Unlock() + conntrack.delete(fp, c) return false } @@ -557,77 +586,35 @@ func (f *Firewall) inConns(fp firewall.Packet, h *HostInfo, caPool *cert.CAPool, ) } - c.rulesVersion = f.rulesVersion + conntrack.store(fp, newConn(expires, c.incoming, f.rulesVersion)) + return true } - switch fp.Protocol { - case iputil.IPProtocolTCP: - c.Expires = time.Now().Add(f.TCPTimeout) - case iputil.IPProtocolUDP: - c.Expires = time.Now().Add(f.UDPTimeout) - default: - c.Expires = time.Now().Add(f.DefaultTimeout) - } - - conntrack.Unlock() - - if localCache != nil { - localCache[fp] = struct{}{} + // Most packets land in the same second as the one before, skip writing an unchanged expiry + if c.expires.Load() != expires { + c.expires.Store(expires) } return true } func (f *Firewall) addConn(fp firewall.Packet, incoming bool) { - var timeout time.Duration - c := &conn{} - - switch fp.Protocol { - case iputil.IPProtocolTCP: - timeout = f.TCPTimeout - case iputil.IPProtocolUDP: - timeout = f.UDPTimeout - default: - timeout = f.DefaultTimeout - } - - conntrack := f.Conntrack - conntrack.Lock() - if _, ok := conntrack.Conns[fp]; !ok { - conntrack.TimerWheel.Advance(time.Now()) - conntrack.TimerWheel.Add(fp, timeout) - } - // Record which rulesVersion allowed this connection, so we can retest after // firewall reload - c.incoming = incoming - c.rulesVersion = f.rulesVersion - c.Expires = time.Now().Add(timeout) - conntrack.Conns[fp] = c - conntrack.Unlock() + conntrack := f.Conntrack + conntrack.store(fp, newConn(conntrack.now.Load()+f.timeout(fp.Protocol), incoming, f.rulesVersion)) } -// 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) { - // Are we still tracking this conn? - conntrack := f.Conntrack - t, ok := conntrack.Conns[p] - if !ok { - return +// timeout returns how many seconds a conntrack entry for proto lives past its last packet +func (f *Firewall) timeout(proto uint8) uint32 { + switch proto { + case iputil.IPProtocolTCP: + return uint32(f.TCPTimeout / time.Second) + case iputil.IPProtocolUDP: + return uint32(f.UDPTimeout / time.Second) + default: + return uint32(f.DefaultTimeout / time.Second) } - - newT := t.Expires.Sub(time.Now()) - - // Timeout is in the future, re-add the timer - if newT > 0 { - conntrack.TimerWheel.Advance(time.Now()) - conntrack.TimerWheel.Add(p, newT) - return - } - - // This conn is done - delete(conntrack.Conns, p) } func (ft *FirewallTable) match(p firewall.Packet, incoming bool, c *cert.CachedCertificate, caPool *cert.CAPool) bool { diff --git a/firewall/cache.go b/firewall/cache.go deleted file mode 100644 index 3e34e6ea..00000000 --- a/firewall/cache.go +++ /dev/null @@ -1,69 +0,0 @@ -package firewall - -import ( - "context" - "log/slog" - "sync/atomic" - "time" - - "github.com/slackhq/nebula/logging" -) - -// 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 ConntrackCacheTicker struct { - cacheV uint64 - cacheTick atomic.Uint64 - - l *slog.Logger - cache ConntrackCache -} - -func NewConntrackCacheTicker(ctx context.Context, l *slog.Logger, d time.Duration) *ConntrackCacheTicker { - if d == 0 { - return nil - } - - c := &ConntrackCacheTicker{ - l: l, - cache: ConntrackCache{}, - } - - go c.tick(ctx, d) - - return c -} - -func (c *ConntrackCacheTicker) tick(ctx context.Context, d time.Duration) { - t := time.NewTicker(d) - defer t.Stop() - for { - select { - case <-ctx.Done(): - return - case <-t.C: - c.cacheTick.Add(1) - } - } -} - -// Get checks if the cache ticker has moved to the next version before returning -// the map. If it has moved, we reset the map. -func (c *ConntrackCacheTicker) Get() ConntrackCache { - if c == nil { - return nil - } - if tick := c.cacheTick.Load(); tick != c.cacheV { - c.cacheV = tick - if ll := len(c.cache); ll > 0 { - if c.l.Enabled(context.Background(), logging.LevelTrace) { - c.l.Log(context.Background(), logging.LevelTrace, "resetting conntrack cache", "len", ll) - } - c.cache = make(ConntrackCache, ll) - } - } - - return c.cache -} diff --git a/firewall/cache_test.go b/firewall/cache_test.go deleted file mode 100644 index 2c8ad089..00000000 --- a/firewall/cache_test.go +++ /dev/null @@ -1,70 +0,0 @@ -package firewall - -import ( - "bytes" - "log/slog" - "strings" - "testing" - - "github.com/slackhq/nebula/logging" - "github.com/slackhq/nebula/test" - "github.com/stretchr/testify/assert" -) - -// The tests below pin the log format produced by ConntrackCacheTicker.Get -// so changes cannot silently break what operators are grepping for. The -// ticker's internal state (cache + cacheTick) is poked directly to avoid -// racing a goroutine-driven tick in tests. - -func newFixedTicker(t *testing.T, l *slog.Logger, cacheLen int) *ConntrackCacheTicker { - t.Helper() - c := &ConntrackCacheTicker{ - l: l, - cache: make(ConntrackCache, cacheLen), - } - for i := range cacheLen { - c.cache[Packet{LocalPort: uint16(i) + 1}] = struct{}{} - } - c.cacheTick.Store(1) // cacheV starts at 0, so Get() takes the reset path - return c -} - -func TestConntrackCacheTicker_Get_TextFormat(t *testing.T) { - buf := &bytes.Buffer{} - l := test.NewLoggerWithOutputAndLevel(buf, logging.LevelTrace) - - c := newFixedTicker(t, l, 3) - c.Get() - - assert.Equal(t, "level=DEBUG-4 msg=\"resetting conntrack cache\" len=3\n", buf.String()) -} - -func TestConntrackCacheTicker_Get_JSONFormat(t *testing.T) { - buf := &bytes.Buffer{} - l := test.NewJSONLoggerWithOutput(buf, logging.LevelTrace) - - c := newFixedTicker(t, l, 2) - c.Get() - - assert.JSONEq(t, `{"level":"DEBUG-4","msg":"resetting conntrack cache","len":2}`, strings.TrimSpace(buf.String())) -} - -func TestConntrackCacheTicker_Get_QuietBelowTrace(t *testing.T) { - buf := &bytes.Buffer{} - l := test.NewLoggerWithOutputAndLevel(buf, slog.LevelDebug) - - c := newFixedTicker(t, l, 5) - c.Get() - - assert.Empty(t, buf.String()) -} - -func TestConntrackCacheTicker_Get_QuietWhenCacheEmpty(t *testing.T) { - buf := &bytes.Buffer{} - l := test.NewLoggerWithOutputAndLevel(buf, logging.LevelTrace) - - c := newFixedTicker(t, l, 0) - c.Get() - - assert.Empty(t, buf.String()) -} diff --git a/firewall/packet.go b/firewall/packet.go index 9f2df76f..5f5589fe 100644 --- a/firewall/packet.go +++ b/firewall/packet.go @@ -26,6 +26,10 @@ type Packet struct { RemotePort uint16 Protocol uint8 Fragment bool + + // pad fills what would otherwise be compiler padding. A Packet with no padding hashes as one + // block of memory, which makes it a much faster map key. Always zero. + pad [2]byte } func (fp *Packet) Copy() *Packet { diff --git a/firewall_test.go b/firewall_test.go index b8727915..befe7072 100644 --- a/firewall_test.go +++ b/firewall_test.go @@ -2,6 +2,7 @@ package nebula import ( "bytes" + "context" "errors" "log/slog" "math" @@ -23,39 +24,12 @@ func TestNewFirewall(t *testing.T) { l := test.NewLogger() c := &dummyCert{} fw := NewFirewall(l, time.Second, time.Minute, time.Hour, c) - conntrack := fw.Conntrack - assert.NotNil(t, conntrack) - assert.NotNil(t, conntrack.Conns) - assert.NotNil(t, conntrack.TimerWheel) + assert.NotNil(t, fw.Conntrack) assert.NotNil(t, fw.InRules) assert.NotNil(t, fw.OutRules) assert.Equal(t, time.Second, fw.TCPTimeout) assert.Equal(t, time.Minute, fw.UDPTimeout) assert.Equal(t, time.Hour, fw.DefaultTimeout) - - assert.Equal(t, time.Hour, conntrack.TimerWheel.wheelDuration) - assert.Equal(t, time.Hour, conntrack.TimerWheel.wheelDuration) - assert.Equal(t, 3602, conntrack.TimerWheel.wheelLen) - - fw = NewFirewall(l, time.Second, time.Hour, time.Minute, c) - assert.Equal(t, time.Hour, conntrack.TimerWheel.wheelDuration) - assert.Equal(t, 3602, conntrack.TimerWheel.wheelLen) - - fw = NewFirewall(l, time.Hour, time.Second, time.Minute, c) - assert.Equal(t, time.Hour, conntrack.TimerWheel.wheelDuration) - assert.Equal(t, 3602, conntrack.TimerWheel.wheelLen) - - fw = NewFirewall(l, time.Hour, time.Minute, time.Second, c) - assert.Equal(t, time.Hour, conntrack.TimerWheel.wheelDuration) - assert.Equal(t, 3602, conntrack.TimerWheel.wheelLen) - - fw = NewFirewall(l, time.Minute, time.Hour, time.Second, c) - assert.Equal(t, time.Hour, conntrack.TimerWheel.wheelDuration) - assert.Equal(t, 3602, conntrack.TimerWheel.wheelLen) - - fw = NewFirewall(l, time.Minute, time.Second, time.Hour, c) - assert.Equal(t, time.Hour, conntrack.TimerWheel.wheelDuration) - assert.Equal(t, 3602, conntrack.TimerWheel.wheelLen) } func TestFirewall_AddRule(t *testing.T) { @@ -212,44 +186,44 @@ func TestFirewall_Drop(t *testing.T) { cp := cert.NewCAPool() // Drop outbound - assert.Equal(t, ErrNoMatchingRule, fw.Drop(p, false, &h, cp, nil)) + assert.Equal(t, ErrNoMatchingRule, fw.Drop(p, false, &h, cp)) // Allow inbound resetConntrack(fw) - require.NoError(t, fw.Drop(p, true, &h, cp, nil)) + require.NoError(t, fw.Drop(p, true, &h, cp)) // Allow outbound because conntrack - require.NoError(t, fw.Drop(p, false, &h, cp, nil)) + require.NoError(t, fw.Drop(p, false, &h, cp)) // test remote mismatch oldRemote := p.RemoteAddr p.RemoteAddr = netip.MustParseAddr("1.2.3.10") - assert.Equal(t, fw.Drop(p, false, &h, cp, nil), ErrInvalidRemoteIP) + assert.Equal(t, fw.Drop(p, false, &h, cp), ErrInvalidRemoteIP) p.RemoteAddr = oldRemote // ensure signer doesn't get in the way of group checks fw = NewFirewall(l, time.Second, time.Minute, time.Hour, &c) require.NoError(t, fw.AddRule(true, firewall.ProtoAny, 0, 0, []string{"nope"}, "", "", "", "", "signer-shasum")) require.NoError(t, fw.AddRule(true, firewall.ProtoAny, 0, 0, []string{"default-group"}, "", "", "", "", "signer-shasum-bad")) - assert.Equal(t, fw.Drop(p, true, &h, cp, nil), ErrNoMatchingRule) + assert.Equal(t, fw.Drop(p, true, &h, cp), ErrNoMatchingRule) // test caSha doesn't drop on match fw = NewFirewall(l, time.Second, time.Minute, time.Hour, &c) require.NoError(t, fw.AddRule(true, firewall.ProtoAny, 0, 0, []string{"nope"}, "", "", "", "", "signer-shasum-bad")) require.NoError(t, fw.AddRule(true, firewall.ProtoAny, 0, 0, []string{"default-group"}, "", "", "", "", "signer-shasum")) - require.NoError(t, fw.Drop(p, true, &h, cp, nil)) + require.NoError(t, fw.Drop(p, true, &h, cp)) // ensure ca name doesn't get in the way of group checks cp.CAs["signer-shasum"] = &cert.CachedCertificate{Certificate: &dummyCert{name: "ca-good"}} fw = NewFirewall(l, time.Second, time.Minute, time.Hour, &c) require.NoError(t, fw.AddRule(true, firewall.ProtoAny, 0, 0, []string{"nope"}, "", "", "", "ca-good", "")) require.NoError(t, fw.AddRule(true, firewall.ProtoAny, 0, 0, []string{"default-group"}, "", "", "", "ca-good-bad", "")) - assert.Equal(t, fw.Drop(p, true, &h, cp, nil), ErrNoMatchingRule) + assert.Equal(t, fw.Drop(p, true, &h, cp), ErrNoMatchingRule) // test caName doesn't drop on match cp.CAs["signer-shasum"] = &cert.CachedCertificate{Certificate: &dummyCert{name: "ca-good"}} fw = NewFirewall(l, time.Second, time.Minute, time.Hour, &c) require.NoError(t, fw.AddRule(true, firewall.ProtoAny, 0, 0, []string{"nope"}, "", "", "", "ca-good-bad", "")) require.NoError(t, fw.AddRule(true, firewall.ProtoAny, 0, 0, []string{"default-group"}, "", "", "", "ca-good", "")) - require.NoError(t, fw.Drop(p, true, &h, cp, nil)) + require.NoError(t, fw.Drop(p, true, &h, cp)) } func TestFirewall_DropV6(t *testing.T) { @@ -290,44 +264,44 @@ func TestFirewall_DropV6(t *testing.T) { cp := cert.NewCAPool() // Drop outbound - assert.Equal(t, ErrNoMatchingRule, fw.Drop(p, false, &h, cp, nil)) + assert.Equal(t, ErrNoMatchingRule, fw.Drop(p, false, &h, cp)) // Allow inbound resetConntrack(fw) - require.NoError(t, fw.Drop(p, true, &h, cp, nil)) + require.NoError(t, fw.Drop(p, true, &h, cp)) // Allow outbound because conntrack - require.NoError(t, fw.Drop(p, false, &h, cp, nil)) + require.NoError(t, fw.Drop(p, false, &h, cp)) // test remote mismatch oldRemote := p.RemoteAddr p.RemoteAddr = netip.MustParseAddr("fd12::56") - assert.Equal(t, fw.Drop(p, false, &h, cp, nil), ErrInvalidRemoteIP) + assert.Equal(t, fw.Drop(p, false, &h, cp), ErrInvalidRemoteIP) p.RemoteAddr = oldRemote // ensure signer doesn't get in the way of group checks fw = NewFirewall(l, time.Second, time.Minute, time.Hour, &c) require.NoError(t, fw.AddRule(true, firewall.ProtoAny, 0, 0, []string{"nope"}, "", "", "", "", "signer-shasum")) require.NoError(t, fw.AddRule(true, firewall.ProtoAny, 0, 0, []string{"default-group"}, "", "", "", "", "signer-shasum-bad")) - assert.Equal(t, fw.Drop(p, true, &h, cp, nil), ErrNoMatchingRule) + assert.Equal(t, fw.Drop(p, true, &h, cp), ErrNoMatchingRule) // test caSha doesn't drop on match fw = NewFirewall(l, time.Second, time.Minute, time.Hour, &c) require.NoError(t, fw.AddRule(true, firewall.ProtoAny, 0, 0, []string{"nope"}, "", "", "", "", "signer-shasum-bad")) require.NoError(t, fw.AddRule(true, firewall.ProtoAny, 0, 0, []string{"default-group"}, "", "", "", "", "signer-shasum")) - require.NoError(t, fw.Drop(p, true, &h, cp, nil)) + require.NoError(t, fw.Drop(p, true, &h, cp)) // ensure ca name doesn't get in the way of group checks cp.CAs["signer-shasum"] = &cert.CachedCertificate{Certificate: &dummyCert{name: "ca-good"}} fw = NewFirewall(l, time.Second, time.Minute, time.Hour, &c) require.NoError(t, fw.AddRule(true, firewall.ProtoAny, 0, 0, []string{"nope"}, "", "", "", "ca-good", "")) require.NoError(t, fw.AddRule(true, firewall.ProtoAny, 0, 0, []string{"default-group"}, "", "", "", "ca-good-bad", "")) - assert.Equal(t, fw.Drop(p, true, &h, cp, nil), ErrNoMatchingRule) + assert.Equal(t, fw.Drop(p, true, &h, cp), ErrNoMatchingRule) // test caName doesn't drop on match cp.CAs["signer-shasum"] = &cert.CachedCertificate{Certificate: &dummyCert{name: "ca-good"}} fw = NewFirewall(l, time.Second, time.Minute, time.Hour, &c) require.NoError(t, fw.AddRule(true, firewall.ProtoAny, 0, 0, []string{"nope"}, "", "", "", "ca-good-bad", "")) require.NoError(t, fw.AddRule(true, firewall.ProtoAny, 0, 0, []string{"default-group"}, "", "", "", "ca-good", "")) - require.NoError(t, fw.Drop(p, true, &h, cp, nil)) + require.NoError(t, fw.Drop(p, true, &h, cp)) } func BenchmarkFirewallTable_match(b *testing.B) { @@ -534,10 +508,10 @@ func TestFirewall_Drop2(t *testing.T) { cp := cert.NewCAPool() // h1/c1 lacks the proper groups - require.ErrorIs(t, fw.Drop(p, true, &h1, cp, nil), ErrNoMatchingRule) + require.ErrorIs(t, fw.Drop(p, true, &h1, cp), ErrNoMatchingRule) // c has the proper groups resetConntrack(fw) - require.NoError(t, fw.Drop(p, true, &h, cp, nil)) + require.NoError(t, fw.Drop(p, true, &h, cp)) } func TestFirewall_Drop3(t *testing.T) { @@ -614,18 +588,18 @@ func TestFirewall_Drop3(t *testing.T) { cp := cert.NewCAPool() // c1 should pass because host match - require.NoError(t, fw.Drop(p, true, &h1, cp, nil)) + require.NoError(t, fw.Drop(p, true, &h1, cp)) // c2 should pass because ca sha match resetConntrack(fw) - require.NoError(t, fw.Drop(p, true, &h2, cp, nil)) + require.NoError(t, fw.Drop(p, true, &h2, cp)) // c3 should fail because no match resetConntrack(fw) - assert.Equal(t, fw.Drop(p, true, &h3, cp, nil), ErrNoMatchingRule) + assert.Equal(t, fw.Drop(p, true, &h3, cp), ErrNoMatchingRule) // Test a remote address match fw = NewFirewall(l, time.Second, time.Minute, time.Hour, c.Certificate) require.NoError(t, fw.AddRule(true, firewall.ProtoAny, 1, 1, []string{}, "", "1.2.3.4/24", "", "", "")) - require.NoError(t, fw.Drop(p, true, &h1, cp, nil)) + require.NoError(t, fw.Drop(p, true, &h1, cp)) } func TestFirewall_Drop3V6(t *testing.T) { @@ -662,7 +636,7 @@ func TestFirewall_Drop3V6(t *testing.T) { fw := NewFirewall(l, time.Second, time.Minute, time.Hour, c.Certificate) cp := cert.NewCAPool() require.NoError(t, fw.AddRule(true, firewall.ProtoAny, 1, 1, []string{}, "", "fd12::34/120", "", "", "")) - require.NoError(t, fw.Drop(p, true, &h, cp, nil)) + require.NoError(t, fw.Drop(p, true, &h, cp)) } func TestFirewall_DropConntrackReload(t *testing.T) { @@ -703,12 +677,12 @@ func TestFirewall_DropConntrackReload(t *testing.T) { cp := cert.NewCAPool() // Drop outbound - assert.Equal(t, fw.Drop(p, false, &h, cp, nil), ErrNoMatchingRule) + assert.Equal(t, fw.Drop(p, false, &h, cp), ErrNoMatchingRule) // Allow inbound resetConntrack(fw) - require.NoError(t, fw.Drop(p, true, &h, cp, nil)) + require.NoError(t, fw.Drop(p, true, &h, cp)) // Allow outbound because conntrack - require.NoError(t, fw.Drop(p, false, &h, cp, nil)) + require.NoError(t, fw.Drop(p, false, &h, cp)) oldFw := fw fw = NewFirewall(l, time.Second, time.Minute, time.Hour, c.Certificate) @@ -717,7 +691,7 @@ func TestFirewall_DropConntrackReload(t *testing.T) { fw.rulesVersion = oldFw.rulesVersion + 1 // Allow outbound because conntrack and new rules allow port 10 - require.NoError(t, fw.Drop(p, false, &h, cp, nil)) + require.NoError(t, fw.Drop(p, false, &h, cp)) oldFw = fw fw = NewFirewall(l, time.Second, time.Minute, time.Hour, c.Certificate) @@ -726,7 +700,7 @@ func TestFirewall_DropConntrackReload(t *testing.T) { fw.rulesVersion = oldFw.rulesVersion + 1 // Drop outbound because conntrack doesn't match new ruleset - assert.Equal(t, fw.Drop(p, false, &h, cp, nil), ErrNoMatchingRule) + assert.Equal(t, fw.Drop(p, false, &h, cp), ErrNoMatchingRule) } func TestFirewall_ICMPPortBehavior(t *testing.T) { @@ -771,12 +745,12 @@ func TestFirewall_ICMPPortBehavior(t *testing.T) { p.LocalPort = 0 p.RemotePort = 0 // Drop outbound - assert.Equal(t, fw.Drop(*p, false, &h, cp, nil), ErrNoMatchingRule) + assert.Equal(t, fw.Drop(*p, false, &h, cp), ErrNoMatchingRule) // Allow inbound resetConntrack(fw) - require.NoError(t, fw.Drop(*p, true, &h, cp, nil)) + require.NoError(t, fw.Drop(*p, true, &h, cp)) //now also allow outbound - require.NoError(t, fw.Drop(*p, false, &h, cp, nil)) + require.NoError(t, fw.Drop(*p, false, &h, cp)) }) t.Run("nonzero ports", func(t *testing.T) { @@ -784,12 +758,12 @@ func TestFirewall_ICMPPortBehavior(t *testing.T) { p.LocalPort = 0xabcd p.RemotePort = 0x1234 // Drop outbound - assert.Equal(t, fw.Drop(*p, false, &h, cp, nil), ErrNoMatchingRule) + assert.Equal(t, fw.Drop(*p, false, &h, cp), ErrNoMatchingRule) // Allow inbound resetConntrack(fw) - require.NoError(t, fw.Drop(*p, true, &h, cp, nil)) + require.NoError(t, fw.Drop(*p, true, &h, cp)) //now also allow outbound - require.NoError(t, fw.Drop(*p, false, &h, cp, nil)) + require.NoError(t, fw.Drop(*p, false, &h, cp)) }) }) @@ -801,12 +775,12 @@ func TestFirewall_ICMPPortBehavior(t *testing.T) { p.LocalPort = 0 p.RemotePort = 0 // Drop outbound - assert.Equal(t, fw.Drop(*p, false, &h, cp, nil), ErrNoMatchingRule) + assert.Equal(t, fw.Drop(*p, false, &h, cp), ErrNoMatchingRule) // Allow inbound resetConntrack(fw) - assert.Equal(t, fw.Drop(*p, true, &h, cp, nil), ErrNoMatchingRule) + assert.Equal(t, fw.Drop(*p, true, &h, cp), ErrNoMatchingRule) //now also allow outbound - assert.Equal(t, fw.Drop(*p, false, &h, cp, nil), ErrNoMatchingRule) + assert.Equal(t, fw.Drop(*p, false, &h, cp), ErrNoMatchingRule) }) t.Run("nonzero ports, still blocked", func(t *testing.T) { @@ -814,12 +788,12 @@ func TestFirewall_ICMPPortBehavior(t *testing.T) { p.LocalPort = 0xabcd p.RemotePort = 0x1234 // Drop outbound - assert.Equal(t, fw.Drop(*p, false, &h, cp, nil), ErrNoMatchingRule) + assert.Equal(t, fw.Drop(*p, false, &h, cp), ErrNoMatchingRule) // Allow inbound resetConntrack(fw) - assert.Equal(t, fw.Drop(*p, true, &h, cp, nil), ErrNoMatchingRule) + assert.Equal(t, fw.Drop(*p, true, &h, cp), ErrNoMatchingRule) //now also allow outbound - assert.Equal(t, fw.Drop(*p, false, &h, cp, nil), ErrNoMatchingRule) + assert.Equal(t, fw.Drop(*p, false, &h, cp), ErrNoMatchingRule) }) t.Run("nonzero, matching ports, still blocked", func(t *testing.T) { @@ -827,12 +801,12 @@ func TestFirewall_ICMPPortBehavior(t *testing.T) { p.LocalPort = 80 p.RemotePort = 80 // Drop outbound - assert.Equal(t, fw.Drop(*p, false, &h, cp, nil), ErrNoMatchingRule) + assert.Equal(t, fw.Drop(*p, false, &h, cp), ErrNoMatchingRule) // Allow inbound resetConntrack(fw) - assert.Equal(t, fw.Drop(*p, true, &h, cp, nil), ErrNoMatchingRule) + assert.Equal(t, fw.Drop(*p, true, &h, cp), ErrNoMatchingRule) //now also allow outbound - assert.Equal(t, fw.Drop(*p, false, &h, cp, nil), ErrNoMatchingRule) + assert.Equal(t, fw.Drop(*p, false, &h, cp), ErrNoMatchingRule) }) }) t.Run("Any proto, any port", func(t *testing.T) { @@ -844,12 +818,12 @@ func TestFirewall_ICMPPortBehavior(t *testing.T) { p.LocalPort = 0 p.RemotePort = 0 // Drop outbound - assert.Equal(t, fw.Drop(*p, false, &h, cp, nil), ErrNoMatchingRule) + assert.Equal(t, fw.Drop(*p, false, &h, cp), ErrNoMatchingRule) // Allow inbound resetConntrack(fw) - require.NoError(t, fw.Drop(*p, true, &h, cp, nil)) + require.NoError(t, fw.Drop(*p, true, &h, cp)) //now also allow outbound - require.NoError(t, fw.Drop(*p, false, &h, cp, nil)) + require.NoError(t, fw.Drop(*p, false, &h, cp)) }) t.Run("nonzero ports, allowed", func(t *testing.T) { @@ -858,15 +832,15 @@ func TestFirewall_ICMPPortBehavior(t *testing.T) { p.LocalPort = 0xabcd p.RemotePort = 0x1234 // Drop outbound - assert.Equal(t, fw.Drop(*p, false, &h, cp, nil), ErrNoMatchingRule) + assert.Equal(t, fw.Drop(*p, false, &h, cp), ErrNoMatchingRule) // Allow inbound resetConntrack(fw) - require.NoError(t, fw.Drop(*p, true, &h, cp, nil)) + require.NoError(t, fw.Drop(*p, true, &h, cp)) //now also allow outbound - require.NoError(t, fw.Drop(*p, false, &h, cp, nil)) + require.NoError(t, fw.Drop(*p, false, &h, cp)) //different ID is blocked p.RemotePort++ - require.Equal(t, fw.Drop(*p, false, &h, cp, nil), ErrNoMatchingRule) + require.Equal(t, fw.Drop(*p, false, &h, cp), ErrNoMatchingRule) }) }) @@ -914,7 +888,7 @@ func TestFirewall_DropIPSpoofing(t *testing.T) { Protocol: iputil.IPProtocolUDP, Fragment: false, } - assert.Equal(t, fw.Drop(p, true, &h1, cp, nil), ErrInvalidRemoteIP) + assert.Equal(t, fw.Drop(p, true, &h1, cp), ErrInvalidRemoteIP) } func TestFirewall_ConntrackSourceSpoofingAcrossPeers(t *testing.T) { @@ -965,20 +939,84 @@ func TestFirewall_ConntrackSourceSpoofingAcrossPeers(t *testing.T) { Protocol: iputil.IPProtocolUDP, } - require.NoError(t, fw.Drop(flow, true, &victimHI, cp, nil), + require.NoError(t, fw.Drop(flow, true, &victimHI, cp), "victim's own traffic from its own overlay IP must be allowed") unseen := flow unseen.RemotePort = 55001 - assert.Equal(t, ErrInvalidRemoteIP, fw.Drop(unseen, true, &attackerHI, cp, nil), + assert.Equal(t, ErrInvalidRemoteIP, fw.Drop(unseen, true, &attackerHI, cp), "sanity: attacker forging victim's source IP must be rejected when no conntrack entry exists") - got := fw.Drop(flow, true, &attackerHI, cp, nil) + got := fw.Drop(flow, true, &attackerHI, cp) t.Logf("attacker replaying victim's 4-tuple: Drop returned %v (nil == packet ALLOWED == spoof succeeded)", got) assert.Equal(t, ErrInvalidRemoteIP, got, "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 TestFirewallConntrack_sweep(t *testing.T) { + ct := newFirewallConntrack() + live := firewall.Packet{LocalPort: 1} + dead := firewall.Packet{LocalPort: 2} + ct.store(live, newConn(11, true, 0)) + ct.store(dead, newConn(10, true, 0)) + assert.Equal(t, int64(2), ct.count.Load()) + + ct.sweep(10) + _, ok := ct.conns.Load(live) + assert.True(t, ok) + _, ok = ct.conns.Load(dead) + assert.False(t, ok) + assert.Equal(t, int64(1), ct.count.Load()) +} + +func TestFirewallConntrack_storeAndDeleteCount(t *testing.T) { + ct := newFirewallConntrack() + fp := firewall.Packet{LocalPort: 1} + first := newConn(10, true, 0) + ct.store(fp, first) + + // Replacing an entry doesn't add to the count + second := newConn(20, true, 1) + ct.store(fp, second) + assert.Equal(t, int64(1), ct.count.Load()) + + // Deleting a stale entry is a no-op, the newer one stays + ct.delete(fp, first) + v, ok := ct.conns.Load(fp) + assert.True(t, ok) + assert.Same(t, second, v) + assert.Equal(t, int64(1), ct.count.Load()) + + ct.delete(fp, second) + _, ok = ct.conns.Load(fp) + assert.False(t, ok) + assert.Equal(t, int64(0), ct.count.Load()) + + ct.store(fp, first) + ct.store(firewall.Packet{LocalPort: 2}, newConn(10, true, 0)) + ct.clear() + assert.Equal(t, int64(0), ct.count.Load()) +} + +func TestFirewallConntrack_Run(t *testing.T) { + ct := newFirewallConntrack() + ct.epoch = time.Now().Add(-time.Hour) + ct.store(firewall.Packet{LocalPort: 1}, newConn(10, true, 0)) + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + ct.Run(ctx) + close(done) + }() + + // The first tick sets the clock to an hour in and sweeps the long expired entry + require.Eventually(t, func() bool { return ct.count.Load() == 0 }, 5*time.Second, 10*time.Millisecond) + assert.GreaterOrEqual(t, ct.now.Load(), uint32(3600)) + cancel() + <-done +} + // 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 @@ -987,8 +1025,6 @@ func TestFirewall_ConntrackSourceSpoofingAcrossPeers(t *testing.T) { // so the remote-address check is a single netip.Addr compare. // - complex: peer cert has unsafe networks (h.networks populated), so the // remote-address check is a BART lookup. -// - noCache/localCache: whether a per-batch ConntrackCache is supplied, which in -// the original code let the fast path skip straight past the address checks. func BenchmarkFirewallDropConntrackHit(b *testing.B) { l := test.NewLoggerWithOutput(&bytes.Buffer{}) @@ -1036,14 +1072,11 @@ func BenchmarkFirewallDropConntrackHit(b *testing.B) { } cases := []struct { - name string - host *HostInfo - useCache bool + name string + host *HostInfo }{ - {"simple/noCache", simpleHost, false}, - {"simple/localCache", simpleHost, true}, - {"complex/noCache", complexHost, false}, - {"complex/localCache", complexHost, true}, + {"simple", simpleHost}, + {"complex", complexHost}, } for _, tc := range cases { @@ -1052,17 +1085,12 @@ func BenchmarkFirewallDropConntrackHit(b *testing.B) { require.NoError(b, fw.AddRule(true, firewall.ProtoAny, 0, 0, []string{"any"}, "", "", "", "", "")) // Establish the conntrack entry so every benchmarked Drop is a hit. - require.NoError(b, fw.Drop(flow, true, tc.host, cp, nil)) - - var cache firewall.ConntrackCache - if tc.useCache { - cache = firewall.ConntrackCache{} - } + require.NoError(b, fw.Drop(flow, true, tc.host, cp)) b.ReportAllocs() b.ResetTimer() for i := 0; i < b.N; i++ { - if err := fw.Drop(flow, true, tc.host, cp, cache); err != nil { + if err := fw.Drop(flow, true, tc.host, cp); err != nil { b.Fatal(err) } } @@ -1550,7 +1578,7 @@ func (c *testcase) Test(t *testing.T, fw *Firewall) { t.Helper() cp := cert.NewCAPool() resetConntrack(fw) - err := fw.Drop(c.p, true, c.h, cp, nil) + err := fw.Drop(c.p, true, c.h, cp) if c.err == nil { require.NoError(t, err, "failed to not drop remote address %s", c.p.RemoteAddr) } else { @@ -1741,7 +1769,5 @@ 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.Unlock() + fw.Conntrack.clear() } diff --git a/inside.go b/inside.go index 8eedbc49..0089bec1 100644 --- a/inside.go +++ b/inside.go @@ -16,7 +16,7 @@ import ( "github.com/slackhq/nebula/routing" ) -func (f *Interface) consumeInsidePacket(pkt tio.Packet, fwPacket *firewall.ParsedPacket, nb []byte, sendBatch *batch.SendBatch, rejectBuf []byte, q int, localCache firewall.ConntrackCache) { +func (f *Interface) consumeInsidePacket(pkt tio.Packet, fwPacket *firewall.ParsedPacket, nb []byte, sendBatch *batch.SendBatch, rejectBuf []byte, q int) { // borrowed: pkt.Bytes is owned by the originating tio.Queue and is // only valid until the next Read on that queue. Every consumer below // (parse, self-forward, handshake cache, sendInsideMessage) reads it @@ -109,7 +109,7 @@ func (f *Interface) consumeInsidePacket(pkt tio.Packet, fwPacket *firewall.Parse return } - dropReason := f.firewall.Drop(fwPacket.Packet, false, hostinfo, f.pki.GetCAPool(), localCache) + dropReason := f.firewall.Drop(fwPacket.Packet, false, hostinfo, f.pki.GetCAPool()) if dropReason == nil { f.sendInsideMessage(hostinfo, pkt, nb, sendBatch) } else { @@ -380,7 +380,7 @@ func (f *Interface) sendMessageNow(t header.MessageType, st header.MessageSubTyp } // check if packet is in outbound fw rules - dropReason := f.firewall.Drop(fp.Packet, false, hostinfo, f.pki.GetCAPool(), nil) + dropReason := f.firewall.Drop(fp.Packet, false, hostinfo, f.pki.GetCAPool()) if dropReason != nil { if f.l.Enabled(context.Background(), slog.LevelDebug) { f.l.Debug("dropping cached packet", diff --git a/inside_test.go b/inside_test.go index 87952201..2aa40a17 100644 --- a/inside_test.go +++ b/inside_test.go @@ -50,7 +50,7 @@ func newSelfForwardInterface(myAddrs ...netip.Addr) (*Interface, *capturingTun) } func consumeInside(f *Interface, packet []byte) { - f.consumeInsidePacket(tio.Packet{Bytes: packet}, &firewall.ParsedPacket{}, make([]byte, 12), nil, make([]byte, mtu), 0, nil) + f.consumeInsidePacket(tio.Packet{Bytes: packet}, &firewall.ParsedPacket{}, make([]byte, 12), nil, make([]byte, mtu), 0) } // l4Proto describes one upper-layer header for these tests: its IP next-header diff --git a/interface.go b/interface.go index 1e0e3db8..69928165 100644 --- a/interface.go +++ b/interface.go @@ -52,8 +52,6 @@ type InterfaceConfig struct { reQueryEvery uint32 reQueryWait time.Duration - ConntrackCacheTimeout time.Duration - // CpuAffinity, when non-empty, names the CPUs each TUN reader goroutine // should pin to. Queue i pins to CpuAffinity[i % len(CpuAffinity)] — // shorter lists than `routines` cycle. Empty list keeps the default @@ -111,8 +109,6 @@ type Interface struct { rebindEpoch atomic.Uint32 version string - conntrackCacheTimeout time.Duration - ctx context.Context writers []udp.Conn queues []tio.Queue @@ -227,7 +223,6 @@ func NewInterface(ctx context.Context, c *InterfaceConfig) (*Interface, error) { myBroadcastAddrsTable: cs.myVpnBroadcastAddrsTable, relayManager: c.relayManager, connectionManager: c.connectionManager, - conntrackCacheTimeout: c.ConntrackCacheTimeout, cpuAffinity: c.CpuAffinity, pinThreads: c.PinThreads, @@ -362,7 +357,6 @@ type rxContext struct { fwPacket *firewall.ParsedPacket hostmapCache map[uint32]*HostInfo lhh *LightHouseHandler - ctCache *firewall.ConntrackCacheTicker } func newRxContext(f *Interface, q int) *rxContext { @@ -374,7 +368,6 @@ func newRxContext(f *Interface, q int) *rxContext { fwPacket: &firewall.ParsedPacket{}, hostmapCache: map[uint32]*HostInfo{}, lhh: f.lightHouse.NewRequestHandler(), - ctCache: firewall.NewConntrackCacheTicker(f.ctx, f.l, f.conntrackCacheTimeout), } } @@ -444,8 +437,6 @@ func (f *Interface) listenIn(queue tio.Queue, i int) { fwPacket := &firewall.ParsedPacket{} nb := make([]byte, 12, 12) - conntrackCache := firewall.NewConntrackCacheTicker(f.ctx, f.l, f.conntrackCacheTimeout) - for { pkts, err := queue.Read() if err != nil { @@ -458,7 +449,7 @@ func (f *Interface) listenIn(queue tio.Queue, i int) { } for _, pkt := range pkts { - f.consumeInsidePacket(pkt, fwPacket, nb, sb, rejectBuf, i, conntrackCache.Get()) + f.consumeInsidePacket(pkt, fwPacket, nb, sb, rejectBuf, i) // Flush incrementally once a full sendmmsg batch has // accumulated so the first packets of a deep read drain // hit the wire while the rest are still being encrypted. @@ -531,9 +522,8 @@ func (f *Interface) reloadFirewall(c *config.C) { } oldFw := f.firewall - conntrack := oldFw.Conntrack - conntrack.Lock() - defer conntrack.Unlock() + // Conntrack lives as long as the interface, keep using the running one + fw.Conntrack = oldFw.Conntrack fw.rulesVersion = oldFw.rulesVersion + 1 // If rulesVersion is back to zero, we have wrapped all the way around. Be @@ -544,8 +534,7 @@ func (f *Interface) reloadFirewall(c *config.C) { "oldFirewallHashes", oldFw.GetRuleHashes(), "rulesVersion", fw.rulesVersion, ) - } else { - fw.Conntrack = conntrack + fw.Conntrack.clear() } f.firewall = fw diff --git a/interface_emit_test.go b/interface_emit_test.go index b0a9d025..2890a574 100644 --- a/interface_emit_test.go +++ b/interface_emit_test.go @@ -10,7 +10,6 @@ import ( "github.com/rcrowley/go-metrics" "github.com/slackhq/nebula/cert" - "github.com/slackhq/nebula/firewall" "github.com/slackhq/nebula/overlay/overlaytest" "github.com/slackhq/nebula/test" "github.com/slackhq/nebula/udp" @@ -43,7 +42,7 @@ func Test_emitStats_primesGauges(t *testing.T) { hostMap: hostMap, inside: &overlaytest.NoopTun{}, outside: &udp.NoopConn{}, - firewall: &Firewall{Conntrack: &FirewallConntrack{Conns: map[firewall.Packet]*conn{}}}, + firewall: &Firewall{Conntrack: newFirewallConntrack()}, lightHouse: lh, pki: &PKI{}, handshakeManager: NewHandshakeManager(l, hostMap, lh, &udp.NoopConn{}, defaultHandshakeConfig), diff --git a/main.go b/main.go index 52f7bea8..41d6f51b 100644 --- a/main.go +++ b/main.go @@ -113,18 +113,6 @@ func Main(c *config.C, configTest bool, buildVersion string, l *slog.Logger, dev l.Info("Using multiple routines", "routines", routines) } - // EXPERIMENTAL - // Intentionally not documented yet while we do more testing and determine - // a good default value. - conntrackCacheTimeout := c.GetDuration("firewall.conntrack.routine_cache_timeout", 0) - if routines > 1 && !c.IsSet("firewall.conntrack.routine_cache_timeout") { - // Use a different default if we are running with multiple routines - conntrackCacheTimeout = 1 * time.Second - } - if conntrackCacheTimeout > 0 { - l.Info("Using routine-local conntrack cache", "duration", conntrackCacheTimeout) - } - var tun overlay.Device if !configTest { c.CatchHUP(ctx) @@ -274,29 +262,28 @@ func Main(c *config.C, configTest bool, buildVersion string, l *slog.Logger, dev } ifConfig := &InterfaceConfig{ - HostMap: hostMap, - Inside: tun, - Outside: udpConns[0], - pki: pki, - Firewall: fw, - DnsServer: ds, - HandshakeManager: handshakeManager, - connectionManager: connManager, - lightHouse: lightHouse, - tryPromoteEvery: c.GetUint32("counters.try_promote", defaultPromoteEvery), - reQueryEvery: c.GetUint32("counters.requery_every_packets", defaultReQueryEvery), - reQueryWait: c.GetDuration("timers.requery_wait_duration", defaultReQueryWait), - DropLocalBroadcast: c.GetBool("tun.drop_local_broadcast", false), - DropMulticast: c.GetBool("tun.drop_multicast", false), - routines: routines, - MessageMetrics: messageMetrics, - version: buildVersion, - relayManager: NewRelayManager(ctx, l, hostMap, c), - punchy: punchy, - ConntrackCacheTimeout: conntrackCacheTimeout, - CpuAffinity: cpuAffinity, - PinThreads: pinThreads, - l: l, + HostMap: hostMap, + Inside: tun, + Outside: udpConns[0], + pki: pki, + Firewall: fw, + DnsServer: ds, + HandshakeManager: handshakeManager, + connectionManager: connManager, + lightHouse: lightHouse, + tryPromoteEvery: c.GetUint32("counters.try_promote", defaultPromoteEvery), + reQueryEvery: c.GetUint32("counters.requery_every_packets", defaultReQueryEvery), + reQueryWait: c.GetDuration("timers.requery_wait_duration", defaultReQueryWait), + DropLocalBroadcast: c.GetBool("tun.drop_local_broadcast", false), + DropMulticast: c.GetBool("tun.drop_multicast", false), + routines: routines, + MessageMetrics: messageMetrics, + version: buildVersion, + relayManager: NewRelayManager(ctx, l, hostMap, c), + punchy: punchy, + CpuAffinity: cpuAffinity, + PinThreads: pinThreads, + l: l, } var ifce *Interface @@ -316,6 +303,7 @@ func Main(c *config.C, configTest bool, buildVersion string, l *slog.Logger, dev handshakeManager.f = ifce go handshakeManager.Run(ctx) + go ifce.firewall.Conntrack.Run(ctx) punchy.Start(ctx, ifce, hostMap, lightHouse) } diff --git a/outside.go b/outside.go index c28fe1ea..a3a46393 100644 --- a/outside.go +++ b/outside.go @@ -477,7 +477,7 @@ func (f *Interface) handleOutsideMessagePacket(hostinfo *HostInfo, messageCounte return } - dropReason := f.firewall.Drop(rxc.fwPacket.Packet, true, hostinfo, f.pki.GetCAPool(), rxc.ctCache.Get()) + dropReason := f.firewall.Drop(rxc.fwPacket.Packet, true, hostinfo, f.pki.GetCAPool()) if dropReason != nil { f.rejectOutside(out, hostinfo.ConnectionState, hostinfo, rxc.nb, rxc.scratch, rxc.q) if f.l.Enabled(context.Background(), slog.LevelDebug) {